create policy cache

pkg/authorization/cache/policy_cache.go

@@ -0,0 +1,149 @@
+package cache
+
+import (
+	"errors"
+	"fmt"
+
+	kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
+
+	authorizationapi "github.com/openshift/origin/pkg/authorization/api"
+	policyregistry "github.com/openshift/origin/pkg/authorization/registry/policy"
+	bindingregistry "github.com/openshift/origin/pkg/authorization/registry/policybinding"
+)
+
+// PolicyCache maintains a cache of PolicyRules
+type PolicyCache struct {
+	policyBindingIndexer cache.Indexer
+	policyIndexer        cache.Indexer
+
+	bindingRegistry bindingregistry.Registry
+	policyRegistry  policyregistry.Registry
+
+	keyFunc cache.KeyFunc
+}
+
+// TODO: Eliminate listWatch when this merges upstream: https://github.com/GoogleCloudPlatform/kubernetes/pull/4453
+type listFunc func() (runtime.Object, error)
+type watchFunc func(resourceVersion string) (watch.Interface, error)
+type listWatch struct {
+	listFunc  listFunc
+	watchFunc watchFunc
+}
+
+func (lw *listWatch) List() (runtime.Object, error) {
+	return lw.listFunc()
+}
+
+func (lw *listWatch) Watch(resourceVersion string) (watch.Interface, error) {
+	return lw.watchFunc(resourceVersion)
+}
+
+// NewPolicyCache creates a new PolicyCache.  You cannot use a normal client, because you don't want policy guarding the policy from the authorizer
+func NewPolicyCache(bindingRegistry bindingregistry.Registry, policyRegistry policyregistry.Registry) *PolicyCache {
+	result := &PolicyCache{
+		policyIndexer:        cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{"namespace": cache.MetaNamespaceIndexFunc}),
+		policyBindingIndexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{"namespace": cache.MetaNamespaceIndexFunc}),
+
+		keyFunc: cache.MetaNamespaceKeyFunc,
+
+		bindingRegistry: bindingRegistry,
+		policyRegistry:  policyRegistry,
+	}
+	return result
+}
+
+// Run begins watching and synchronizing the cache
+func (c *PolicyCache) Run() {
+	policyBindingReflector, policyReflector := c.configureReflectors()
+
+	policyBindingReflector.Run()
+	policyReflector.Run()
+}
+
+// RunUntil starts a watch and handles watch events. Will restart the watch if it is closed.
+// RunUntil starts a goroutine and returns immediately. It will exit when stopCh is closed.
+func (c *PolicyCache) RunUntil(bindingStopCh <-chan struct{}, policyStopCh <-chan struct{}) {
+	policyBindingReflector, policyReflector := c.configureReflectors()
+
+	policyBindingReflector.RunUntil(bindingStopCh)
+	policyReflector.RunUntil(policyStopCh)
+}
+
+func (c *PolicyCache) configureReflectors() (*cache.Reflector, *cache.Reflector) {
+	ctx := kapi.WithNamespace(kapi.NewContext(), kapi.NamespaceAll)
+
+	policyBindingReflector := cache.NewReflector(
+		&listWatch{
+			listFunc: func() (runtime.Object, error) {
+				return c.bindingRegistry.ListPolicyBindings(ctx, labels.Everything(), labels.Everything())
+			},
+			watchFunc: func(resourceVersion string) (watch.Interface, error) {
+				return c.bindingRegistry.WatchPolicyBindings(ctx, labels.Everything(), labels.Everything(), resourceVersion)
+			},
+		},
+		&authorizationapi.PolicyBinding{},
+		c.policyBindingIndexer,
+	)
+
+	policyReflector := cache.NewReflector(
+		&listWatch{
+			listFunc: func() (runtime.Object, error) {
+				return c.policyRegistry.ListPolicies(ctx, labels.Everything(), labels.Everything())
+			},
+			watchFunc: func(resourceVersion string) (watch.Interface, error) {
+				return c.policyRegistry.WatchPolicies(ctx, labels.Everything(), labels.Everything(), resourceVersion)
+			},
+		},
+		&authorizationapi.Policy{},
+		c.policyIndexer,
+	)
+
+	return policyBindingReflector, policyReflector
+}
+
+// GetPolicy retrieves a specific policy.  It conforms to rulevalidation.PolicyGetter.
+func (c *PolicyCache) GetPolicy(ctx kapi.Context, name string) (*authorizationapi.Policy, error) {
+	namespace, exists := kapi.NamespaceFrom(ctx)
+	if !exists {
+		return nil, errors.New("no namespace found")
+	}
+
+	keyObj := &authorizationapi.Policy{ObjectMeta: kapi.ObjectMeta{Namespace: namespace, Name: name}}
+	key, _ := c.keyFunc(keyObj)
+
+	policy, exists, err := c.policyIndexer.GetByKey(key)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, fmt.Errorf("%v not found", key)
+	}
+
+	return policy.(*authorizationapi.Policy), nil
+}
+
+// ListPolicyBindings obtains list of policyBindings that match a selector.  It conforms to rulevalidation.BindingLister
+func (c *PolicyCache) ListPolicyBindings(ctx kapi.Context, labels, fields labels.Selector) (*authorizationapi.PolicyBindingList, error) {
+	namespace, exists := kapi.NamespaceFrom(ctx)
+	if !exists {
+		return nil, errors.New("no namespace found")
+	}
+
+	bindings, err := c.policyBindingIndexer.Index("namespace", &authorizationapi.PolicyBinding{ObjectMeta: kapi.ObjectMeta{Namespace: namespace}})
+	if err != nil {
+		return nil, err
+	}
+
+	ret := &authorizationapi.PolicyBindingList{
+		Items: make([]authorizationapi.PolicyBinding, 0, len(bindings)),
+	}
+	for i := range bindings {
+		ret.Items = append(ret.Items, *bindings[i].(*authorizationapi.PolicyBinding))
+	}
+
+	return ret, nil
+}

pkg/authorization/cache/policy_cache_test.go

@@ -0,0 +1,95 @@
+package cache
+
+import (
+	"testing"
+	"time"
+
+	kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	authorizationapi "github.com/openshift/origin/pkg/authorization/api"
+	testregistry "github.com/openshift/origin/pkg/authorization/registry/test"
+)
+
+func TestPolicyGet(t *testing.T) {
+	policyStop := make(chan struct{})
+	bindingStop := make(chan struct{})
+	defer close(policyStop)
+	defer close(bindingStop)
+
+	policyRegistry := testregistry.NewPolicyRegistry(testPolicies(), nil)
+	bindingRegistry := testregistry.NewPolicyBindingRegistry(testBindings(), nil)
+
+	policyCache := NewPolicyCache(bindingRegistry, policyRegistry)
+	policyCache.RunUntil(bindingStop, policyStop)
+
+	testStop := make(chan struct{})
+
+	util.Until(func() {
+		ctx := kapi.WithNamespace(kapi.NewContext(), "mallet")
+		policy, policyErr := policyCache.GetPolicy(ctx, authorizationapi.PolicyName)
+
+		bindings, bindingErr := policyCache.ListPolicyBindings(ctx, labels.Everything(), labels.Everything())
+		if (policyErr == nil) && (bindingErr == nil) && (policy != nil) && (len(bindings.Items) == 1) {
+			close(testStop)
+		}
+
+	}, 1*time.Millisecond, testStop)
+}
+
+func testPolicies() []authorizationapi.Policy {
+	return []authorizationapi.Policy{
+		{
+			ObjectMeta: kapi.ObjectMeta{
+				Name:      authorizationapi.PolicyName,
+				Namespace: "mallet",
+			},
+			Roles: map[string]authorizationapi.Role{},
+		}}
+}
+func testBindings() []authorizationapi.PolicyBinding {
+	return []authorizationapi.PolicyBinding{
+		{
+			ObjectMeta: kapi.ObjectMeta{
+				Name:      "mallet",
+				Namespace: "mallet",
+			},
+			RoleBindings: map[string]authorizationapi.RoleBinding{
+				"projectAdmins": {
+					ObjectMeta: kapi.ObjectMeta{
+						Name:      "projectAdmins",
+						Namespace: "mallet",
+					},
+					RoleRef: kapi.ObjectReference{
+						Name:      "admin",
+						Namespace: "mallet",
+					},
+					Users: util.NewStringSet("Matthew"),
+				},
+				"viewers": {
+					ObjectMeta: kapi.ObjectMeta{
+						Name:      "viewers",
+						Namespace: "mallet",
+					},
+					RoleRef: kapi.ObjectReference{
+						Name:      "view",
+						Namespace: "mallet",
+					},
+					Users: util.NewStringSet("Victor"),
+				},
+				"editors": {
+					ObjectMeta: kapi.ObjectMeta{
+						Name:      "editors",
+						Namespace: "mallet",
+					},
+					RoleRef: kapi.ObjectReference{
+						Name:      "edit",
+						Namespace: "mallet",
+					},
+					Users: util.NewStringSet("Edgar"),
+				},
+			},
+		},
+	}
+}

pkg/authorization/registry/test/policy.go

@@ -34,16 +34,21 @@ func (r *PolicyRegistry) ListPolicies(ctx kapi.Context, labels, fields klabels.S
 	}
 
 	namespace := kapi.NamespaceValue(ctx)
-	if len(namespace) == 0 {
-		return nil, errors.New("invalid request.  Namespace parameter required.")
-	}
-
 	list := make([]authorizationapi.Policy, 0)
-	if namespacedPolicies, ok := r.Policies[namespace]; ok {
-		for _, curr := range namespacedPolicies {
-			list = append(list, curr)
+
+	if namespace == kapi.NamespaceAll {
+		for _, curr := range r.Policies {
+			for _, policy := range curr {
+				list = append(list, policy)
+			}
 		}
 
+	} else {
+		if namespacedPolicies, ok := r.Policies[namespace]; ok {
+			for _, curr := range namespacedPolicies {
+				list = append(list, curr)
+			}
+		}
 	}
 
 	return &authorizationapi.PolicyList{

pkg/authorization/registry/test/policybinding.go

@@ -34,16 +34,21 @@ func (r *PolicyBindingRegistry) ListPolicyBindings(ctx kapi.Context, labels, fie
 	}
 
 	namespace := kapi.NamespaceValue(ctx)
-	if len(namespace) == 0 {
-		return nil, errors.New("invalid request.  Namespace parameter required.")
-	}
-
 	list := make([]authorizationapi.PolicyBinding, 0)
-	if namespacedBindings, ok := r.PolicyBindings[namespace]; ok {
-		for _, curr := range namespacedBindings {
-			list = append(list, curr)
+
+	if namespace == kapi.NamespaceAll {
+		for _, curr := range r.PolicyBindings {
+			for _, binding := range curr {
+				list = append(list, binding)
+			}
 		}
 
+	} else {
+		if namespacedBindings, ok := r.PolicyBindings[namespace]; ok {
+			for _, curr := range namespacedBindings {
+				list = append(list, curr)
+			}
+		}
 	}
 
 	return &authorizationapi.PolicyBindingList{

pkg/cmd/server/origin/master.go

@@ -75,6 +75,7 @@ import (
 
 	authorizationapi "github.com/openshift/origin/pkg/authorization/api"
 	"github.com/openshift/origin/pkg/authorization/authorizer"
+	policycache "github.com/openshift/origin/pkg/authorization/cache"
 	authorizationetcd "github.com/openshift/origin/pkg/authorization/registry/etcd"
 	policyregistry "github.com/openshift/origin/pkg/authorization/registry/policy"
 	policybindingregistry "github.com/openshift/origin/pkg/authorization/registry/policybinding"
@@ -115,6 +116,7 @@ type MasterConfig struct {
 	AuthorizationAttributeBuilder authorizer.AuthorizationAttributeBuilder
 	MasterAuthorizationNamespace  string
 
+	PolicyCache               *policycache.PolicyCache
 	ProjectAuthorizationCache *projectauth.AuthorizationCache
 
 	// Map requests to contexts
@@ -559,6 +561,11 @@ func (c *MasterConfig) RunProjectAuthorizationCache() {
 	c.ProjectAuthorizationCache.Run(period)
 }
 
+// RunPolicyCache starts the policy cache
+func (c *MasterConfig) RunPolicyCache() {
+	c.PolicyCache.Run()
+}
+
 // RunAssetServer starts the asset server for the OpenShift UI.
 func (c *MasterConfig) RunAssetServer() {
 	// TODO use	version.Get().GitCommit as an etag cache header

pkg/cmd/server/start.go

@@ -38,6 +38,7 @@ import (
 	"github.com/openshift/origin/pkg/auth/authenticator/request/unionrequest"
 	"github.com/openshift/origin/pkg/auth/authenticator/request/x509request"
 	"github.com/openshift/origin/pkg/authorization/authorizer"
+	policycache "github.com/openshift/origin/pkg/authorization/cache"
 	authorizationetcd "github.com/openshift/origin/pkg/authorization/registry/etcd"
 	"github.com/openshift/origin/pkg/authorization/rulevalidation"
 	projectauth "github.com/openshift/origin/pkg/project/auth"
@@ -367,11 +368,9 @@ func start(cfg *config, args []string) error {
 
 			EtcdHelper: etcdHelper,
 
-			AdmissionControl:              admit.NewAlwaysAdmit(),
-			Authorizer:                    newAuthorizer(etcdHelper, masterAuthorizationNamespace),
-			AuthorizationAttributeBuilder: newAuthorizationAttributeBuilder(requestContextMapper),
-			MasterAuthorizationNamespace:  masterAuthorizationNamespace,
-			RequestContextMapper:          requestContextMapper,
+			AdmissionControl:             admit.NewAlwaysAdmit(),
+			MasterAuthorizationNamespace: masterAuthorizationNamespace,
+			RequestContextMapper:         requestContextMapper,
 
 			ImageFor: imageResolverFn,
 		}
@@ -502,6 +501,13 @@ func start(cfg *config, args []string) error {
 
 		osmaster.BuildClients()
 
+		authorizationEtcd := authorizationetcd.New(etcdHelper)
+		osmaster.PolicyCache = policycache.NewPolicyCache(authorizationEtcd, authorizationEtcd)
+		osmaster.Authorizer = newAuthorizer(osmaster.PolicyCache, masterAuthorizationNamespace)
+		osmaster.AuthorizationAttributeBuilder = newAuthorizationAttributeBuilder(requestContextMapper)
+		// the policy cache must start before you attempt to start any other components
+		osmaster.RunPolicyCache()
+
 		osmaster.ProjectAuthorizationCache = projectauth.NewAuthorizationCache(
 			projectauth.NewReviewer(osmaster.PolicyClient()),
 			osmaster.KubeClient().Namespaces(),
@@ -653,9 +659,8 @@ func start(cfg *config, args []string) error {
 	return nil
 }
 
-func newAuthorizer(etcdHelper tools.EtcdHelper, masterAuthorizationNamespace string) authorizer.Authorizer {
-	authorizationEtcd := authorizationetcd.New(etcdHelper)
-	authorizer := authorizer.NewAuthorizer(masterAuthorizationNamespace, rulevalidation.NewDefaultRuleResolver(authorizationEtcd, authorizationEtcd))
+func newAuthorizer(policyCache *policycache.PolicyCache, masterAuthorizationNamespace string) authorizer.Authorizer {
+	authorizer := authorizer.NewAuthorizer(masterAuthorizationNamespace, rulevalidation.NewDefaultRuleResolver(policyCache, policyCache))
 	return authorizer
 }