Merge pull request #9204 from deads2k/fix-project-watch

Merged by openshift-bot

Author: OpenShift Bot

pkg/project/auth/cache.go

@@ -105,6 +105,9 @@ func (s *neverSkipSynchronizer) SkipSynchronize(prevState string, versionedObjec
 
 // AuthorizationCache maintains a cache on the set of namespaces a user or group can access.
 type AuthorizationCache struct {
+	// allKnownNamespaces we track all the known namespaces, so we can detect deletes.
+	// TODO remove this in favor of a list/watch mechanism for projects
+	allKnownNamespaces        sets.String
 	namespaceStore            cache.Store
 	namespaceInterface        kclient.NamespaceInterface
 	lastSyncResourceVersioner LastSyncResourceVersioner
@@ -132,6 +135,7 @@ type AuthorizationCache struct {
 // NewAuthorizationCache creates a new AuthorizationCache
 func NewAuthorizationCache(reviewer Reviewer, namespaceInterface kclient.NamespaceInterface, policyClient policyclient.ReadOnlyPolicyClient) *AuthorizationCache {
 	result := &AuthorizationCache{
+		allKnownNamespaces:        sets.String{},
 		namespaceStore:            cache.NewStore(cache.MetaNamespaceKeyFunc),
 		namespaceInterface:        namespaceInterface,
 		lastSyncResourceVersioner: &unchangingLastSyncResourceVersioner{},
@@ -203,7 +207,7 @@ func (ac *AuthorizationCache) RemoveWatcher(watcher CacheWatcher) {
 }
 
 // synchronizeNamespaces synchronizes access over each namespace and returns a set of namespace names that were looked at in last sync
-func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) *sets.String {
+func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) sets.String {
 	namespaceSet := sets.NewString()
 	items := ac.namespaceStore.List()
 	for i := range items {
@@ -217,7 +221,7 @@ func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache
 			utilruntime.HandleError(fmt.Errorf("error synchronizing: %v", err))
 		}
 	}
-	return &namespaceSet
+	return namespaceSet
 }
 
 // synchronizePolicies synchronizes access over each policy
@@ -257,16 +261,20 @@ func (ac *AuthorizationCache) synchronizePolicyBindings(userSubjectRecordStore c
 }
 
 // purgeDeletedNamespaces will remove all namespaces enumerated in a reviewRecordStore that are not in the namespace set
-func purgeDeletedNamespaces(namespaceSet *sets.String, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) {
+func (ac *AuthorizationCache) purgeDeletedNamespaces(oldNamespaces, newNamespaces sets.String, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) {
 	reviewRecordItems := reviewRecordStore.List()
 	for i := range reviewRecordItems {
 		reviewRecord := reviewRecordItems[i].(*reviewRecord)
-		if !namespaceSet.Has(reviewRecord.namespace) {
+		if !newNamespaces.Has(reviewRecord.namespace) {
 			deleteNamespaceFromSubjects(userSubjectRecordStore, reviewRecord.users, reviewRecord.namespace)
 			deleteNamespaceFromSubjects(groupSubjectRecordStore, reviewRecord.groups, reviewRecord.namespace)
 			reviewRecordStore.Delete(reviewRecord)
 		}
 	}
+
+	for namespace := range oldNamespaces.Difference(newNamespaces) {
+		ac.notifyWatchers(namespace, nil, sets.String{}, sets.String{})
+	}
 }
 
 // invalidateCache returns true if there was a change in the cluster namespace that holds cluster policy and policy bindings
@@ -327,17 +335,18 @@ func (ac *AuthorizationCache) synchronize() {
 	}
 
 	// iterate over caches and synchronize our three caches
-	namespaceSet := ac.synchronizeNamespaces(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
+	newKnownNamespaces := ac.synchronizeNamespaces(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
 	ac.synchronizePolicies(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
 	ac.synchronizePolicyBindings(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
-	purgeDeletedNamespaces(namespaceSet, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
+	ac.purgeDeletedNamespaces(ac.allKnownNamespaces, newKnownNamespaces, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore)
 
 	// if we did a full rebuild, now we swap the fully rebuilt cache
 	if invalidateCache {
 		ac.userSubjectRecordStore = userSubjectRecordStore
 		ac.groupSubjectRecordStore = groupSubjectRecordStore
 		ac.reviewRecordStore = reviewRecordStore
 	}
+	ac.allKnownNamespaces = newKnownNamespaces
 
 	// we were able to update our cache since this last observation period
 	ac.lastState = currentState
@@ -486,24 +495,11 @@ func addSubjectsToNamespace(subjectRecordStore cache.Store, subjects []string, n
 	}
 }
 
-func (ac *AuthorizationCache) notifyWatchers(namespace string, exists *reviewRecord, latestUsers, latestGroups sets.String) {
-	existingGroups := sets.String{}
-	existingUsers := sets.String{}
-	if exists != nil {
-		existingGroups = sets.NewString(exists.groups...)
-		existingUsers = sets.NewString(exists.users...)
-	}
-
-	// calculate once to avoid fanning out.
-	removedUsers := existingUsers.Difference(latestUsers)
-	removedGroups := existingGroups.Difference(latestGroups)
-	addedUsers := latestUsers.Difference(existingUsers)
-	addedGroups := latestGroups.Difference(existingGroups)
-
+func (ac *AuthorizationCache) notifyWatchers(namespace string, exists *reviewRecord, users, groups sets.String) {
 	ac.watcherLock.Lock()
 	defer ac.watcherLock.Unlock()
 	for _, watcher := range ac.watchers {
-		watcher.GroupMembershipChanged(namespace, latestUsers, latestGroups, removedUsers, removedGroups, addedUsers, addedGroups)
+		watcher.GroupMembershipChanged(namespace, users, groups)
 	}
 }
 

pkg/project/auth/watch.go

@@ -21,7 +21,7 @@ import (
 type CacheWatcher interface {
 	// GroupMembershipChanged is called serially for all changes for all watchers.  This method MUST NOT BLOCK.
 	// The serial nature makes reasoning about the code easy, but if you block in this method you will doom all watchers.
-	GroupMembershipChanged(namespaceName string, latestUsers, latestGroups, removedUsers, removedGroups, addedUsers, addedGroups sets.String)
+	GroupMembershipChanged(namespaceName string, users, groups sets.String)
 }
 
 type WatchableCache interface {
@@ -106,15 +106,13 @@ func NewUserProjectWatcher(username string, groups []string, projectCache *proje
 	return w
 }
 
-func (w *userProjectWatcher) GroupMembershipChanged(namespaceName string, latestUsers, lastestGroups, removedUsers, removedGroups, addedUsers, addedGroups sets.String) {
-	hasAccess := latestUsers.Has(w.username) || lastestGroups.HasAny(w.groups...)
-	removed := !hasAccess && (removedUsers.Has(w.username) || removedGroups.HasAny(w.groups...))
+func (w *userProjectWatcher) GroupMembershipChanged(namespaceName string, users, groups sets.String) {
+	hasAccess := users.Has(w.username) || groups.HasAny(w.groups...)
+	_, known := w.knownProjects[namespaceName]
 
 	switch {
-	case removed:
-		if _, known := w.knownProjects[namespaceName]; !known {
-			return
-		}
+	// this means that we were removed from the project
+	case !hasAccess && known:
 		delete(w.knownProjects, namespaceName)
 
 		select {

pkg/project/auth/watch_test.go

@@ -59,7 +59,7 @@ func TestFullIncoming(t *testing.T) {
 	watcher.cacheIncoming <- watch.Event{Type: watch.Added}
 
 	// this call should not block and we should see a failure
-	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{}, sets.String{}, sets.String{}, sets.NewString("bob"), sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{})
 	if len(fakeAuthCache.removed) != 1 {
 		t.Errorf("should have removed self")
 	}
@@ -103,7 +103,7 @@ func TestAddModifyDeleteEventsByUser(t *testing.T) {
 	watcher, _ := newTestWatcher("bob", nil, newNamespaces("ns-01")...)
 	go watcher.Watch()
 
-	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{}, sets.String{}, sets.String{}, sets.NewString("bob"), sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{})
 	select {
 	case event := <-watcher.ResultChan():
 		if event.Type != watch.Added {
@@ -117,14 +117,14 @@ func TestAddModifyDeleteEventsByUser(t *testing.T) {
 	}
 
 	// the object didn't change, we shouldn't observe it
-	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{}, sets.String{}, sets.String{}, sets.String{}, sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.NewString("bob"), sets.String{})
 	select {
 	case event := <-watcher.ResultChan():
 		t.Fatalf("unexpected event %v", event)
 	case <-time.After(3 * time.Second):
 	}
 
-	watcher.GroupMembershipChanged("ns-01", sets.NewString("alice"), sets.String{}, sets.NewString("bob"), sets.String{}, sets.String{}, sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.NewString("alice"), sets.String{})
 	select {
 	case event := <-watcher.ResultChan():
 		if event.Type != watch.Deleted {
@@ -142,7 +142,7 @@ func TestAddModifyDeleteEventsByGroup(t *testing.T) {
 	watcher, _ := newTestWatcher("bob", []string{"group-one"}, newNamespaces("ns-01")...)
 	go watcher.Watch()
 
-	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-one"), sets.String{}, sets.String{}, sets.String{}, sets.NewString("group-one"))
+	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-one"))
 	select {
 	case event := <-watcher.ResultChan():
 		if event.Type != watch.Added {
@@ -156,14 +156,14 @@ func TestAddModifyDeleteEventsByGroup(t *testing.T) {
 	}
 
 	// the object didn't change, we shouldn't observe it
-	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-one"), sets.String{}, sets.String{}, sets.String{}, sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-one"))
 	select {
 	case event := <-watcher.ResultChan():
 		t.Fatalf("unexpected event %v", event)
 	case <-time.After(3 * time.Second):
 	}
 
-	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-two"), sets.String{}, sets.NewString("group-one"), sets.String{}, sets.String{})
+	watcher.GroupMembershipChanged("ns-01", sets.String{}, sets.NewString("group-two"))
 	select {
 	case event := <-watcher.ResultChan():
 		if event.Type != watch.Deleted {

test/integration/project_test.go

@@ -156,7 +156,7 @@ func TestProjectMustExist(t *testing.T) {
 
 func TestProjectWatch(t *testing.T) {
 	testutil.RequireEtcd(t)
-	_, clusterAdminKubeConfig, err := testserver.StartTestMasterAPI()
+	_, clusterAdminKubeConfig, err := testserver.StartTestMaster()
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -181,21 +181,9 @@ func TestProjectWatch(t *testing.T) {
 	if _, err := testserver.CreateNewProject(clusterAdminClient, *clusterAdminClientConfig, "ns-01", "bob"); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
+	waitForAdd("ns-01", w, t)
 
-	select {
-	case event := <-w.ResultChan():
-		if event.Type != watch.Added {
-			t.Errorf("expected added, got %v", event)
-		}
-		project := event.Object.(*projectapi.Project)
-		if project.Name != "ns-01" {
-			t.Fatalf("expected %v, got %#v", "ns-01", project)
-		}
-
-	case <-time.After(3 * time.Second):
-		t.Fatalf("timeout")
-	}
-
+	// TEST FOR ADD/REMOVE ACCESS
 	joeClient, err := testserver.CreateNewProject(clusterAdminClient, *clusterAdminClientConfig, "ns-02", "joe")
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -209,58 +197,31 @@ func TestProjectWatch(t *testing.T) {
 	if err := addBob.AddRole(); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
-	// wait for the add
-	for {
-		select {
-		case event := <-w.ResultChan():
-			project := event.Object.(*projectapi.Project)
-			t.Logf("got %#v %#v", event, project)
-			if event.Type == watch.Added && project.Name == "ns-02" {
-				return
-			}
-
-		case <-time.After(3 * time.Second):
-			t.Fatalf("timeout")
-		}
-	}
+	waitForAdd("ns-02", w, t)
 
 	if err := addBob.RemoveRole(); err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
+	waitForDelete("ns-02", w, t)
 
-	// wait for the delete
-	for {
-		select {
-		case event := <-w.ResultChan():
-			project := event.Object.(*projectapi.Project)
-			t.Logf("got %#v %#v", event, project)
-			if event.Type == watch.Deleted && project.Name == "ns-02" {
-				return
-			}
+	// TEST FOR DELETE PROJECT
+	if _, err := testserver.CreateNewProject(clusterAdminClient, *clusterAdminClientConfig, "ns-03", "bob"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	waitForAdd("ns-03", w, t)
 
-		case <-time.After(3 * time.Second):
-			t.Fatalf("timeout")
-		}
+	if err := bobClient.Projects().Delete("ns-03"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
 	}
+	// wait for the delete
+	waitForDelete("ns-03", w, t)
 
 	// test the "start from beginning watch"
 	beginningWatch, err := bobClient.Projects().Watch(kapi.ListOptions{ResourceVersion: "0"})
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
-	select {
-	case event := <-beginningWatch.ResultChan():
-		if event.Type != watch.Added {
-			t.Errorf("expected added, got %v", event)
-		}
-		project := event.Object.(*projectapi.Project)
-		if project.Name != "ns-01" {
-			t.Fatalf("expected %v, got %#v", "ns-01", project)
-		}
-
-	case <-time.After(3 * time.Second):
-		t.Fatalf("timeout")
-	}
+	waitForAdd("ns-01", beginningWatch, t)
 
 	fromNowWatch, err := bobClient.Projects().Watch(kapi.ListOptions{})
 	if err != nil {
@@ -274,3 +235,34 @@ func TestProjectWatch(t *testing.T) {
 	}
 
 }
+
+func waitForDelete(projectName string, w watch.Interface, t *testing.T) {
+	for {
+		select {
+		case event := <-w.ResultChan():
+			project := event.Object.(*projectapi.Project)
+			t.Logf("got %#v %#v", event, project)
+			if event.Type == watch.Deleted && project.Name == projectName {
+				return
+			}
+
+		case <-time.After(30 * time.Second):
+			t.Fatalf("timeout: %v", projectName)
+		}
+	}
+}
+func waitForAdd(projectName string, w watch.Interface, t *testing.T) {
+	for {
+		select {
+		case event := <-w.ResultChan():
+			project := event.Object.(*projectapi.Project)
+			t.Logf("got %#v %#v", event, project)
+			if event.Type == watch.Added && project.Name == projectName {
+				return
+			}
+
+		case <-time.After(30 * time.Second):
+			t.Fatalf("timeout: %v", projectName)
+		}
+	}
+}