Merge pull request #117836 from ruquanzhao/addPodUID

Passing podUID in AddReference and DeleteReference

pkg/kubelet/util/manager/cache_based_manager.go

@@ -22,7 +22,7 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/kubernetes/pkg/kubelet/util"
 
@@ -92,7 +92,7 @@ func isObjectOlder(newObject, oldObject runtime.Object) bool {
 	return newVersion < oldVersion
 }
 
-func (s *objectStore) AddReference(namespace, name string) {
+func (s *objectStore) AddReference(namespace, name string, _ types.UID) {
 	key := objectKey{namespace: namespace, name: name}
 
 	// AddReference is called from RegisterPod, thus it needs to be efficient.
@@ -114,7 +114,7 @@ func (s *objectStore) AddReference(namespace, name string) {
 	item.data = nil
 }
 
-func (s *objectStore) DeleteReference(namespace, name string) {
+func (s *objectStore) DeleteReference(namespace, name string, _ types.UID) {
 	key := objectKey{namespace: namespace, name: name}
 
 	s.lock.Lock()
@@ -225,7 +225,7 @@ func (c *cacheBasedManager) RegisterPod(pod *v1.Pod) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	for name := range names {
-		c.objectStore.AddReference(pod.Namespace, name)
+		c.objectStore.AddReference(pod.Namespace, name, pod.UID)
 	}
 	var prev *v1.Pod
 	key := objectKey{namespace: pod.Namespace, name: pod.Name, uid: pod.UID}
@@ -238,7 +238,7 @@ func (c *cacheBasedManager) RegisterPod(pod *v1.Pod) {
 			// names and prev need to have their ref counts decremented. Any that
 			// are only in prev need to be completely removed. This unconditional
 			// call takes care of both cases.
-			c.objectStore.DeleteReference(prev.Namespace, name)
+			c.objectStore.DeleteReference(prev.Namespace, name, prev.UID)
 		}
 	}
 }
@@ -252,7 +252,7 @@ func (c *cacheBasedManager) UnregisterPod(pod *v1.Pod) {
 	delete(c.registeredPods, key)
 	if prev != nil {
 		for name := range c.getReferencedObjects(prev) {
-			c.objectStore.DeleteReference(prev.Namespace, name)
+			c.objectStore.DeleteReference(prev.Namespace, name, prev.UID)
 		}
 	}
 }

pkg/kubelet/util/manager/cache_based_manager_test.go

@@ -89,13 +89,13 @@ func newCacheBasedSecretManager(store Store) Manager {
 func TestSecretStore(t *testing.T) {
 	fakeClient := &fake.Clientset{}
 	store := newSecretStore(fakeClient, clock.RealClock{}, noObjectTTL, 0)
-	store.AddReference("ns1", "name1")
-	store.AddReference("ns2", "name2")
-	store.AddReference("ns1", "name1")
-	store.AddReference("ns1", "name1")
-	store.DeleteReference("ns1", "name1")
-	store.DeleteReference("ns2", "name2")
-	store.AddReference("ns3", "name3")
+	store.AddReference("ns1", "name1", "pod1")
+	store.AddReference("ns2", "name2", "pod2")
+	store.AddReference("ns1", "name1", "pod3")
+	store.AddReference("ns1", "name1", "pod4")
+	store.DeleteReference("ns1", "name1", "pod1")
+	store.DeleteReference("ns2", "name2", "pod2")
+	store.AddReference("ns3", "name3", "pod5")
 
 	// Adds don't issue Get requests.
 	actions := fakeClient.Actions()
@@ -123,7 +123,7 @@ func TestSecretStore(t *testing.T) {
 func TestSecretStoreDeletingSecret(t *testing.T) {
 	fakeClient := &fake.Clientset{}
 	store := newSecretStore(fakeClient, clock.RealClock{}, noObjectTTL, 0)
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 
 	result := &v1.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name", ResourceVersion: "10"}}
 	fakeClient.AddReactor("get", "secrets", func(action core.Action) (bool, runtime.Object, error) {
@@ -155,7 +155,7 @@ func TestSecretStoreGetAlwaysRefresh(t *testing.T) {
 	store := newSecretStore(fakeClient, fakeClock, noObjectTTL, 0)
 
 	for i := 0; i < 10; i++ {
-		store.AddReference(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i))
+		store.AddReference(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i), types.UID(fmt.Sprintf("pod-%d", i)))
 	}
 	fakeClient.ClearActions()
 
@@ -182,7 +182,7 @@ func TestSecretStoreGetNeverRefresh(t *testing.T) {
 	store := newSecretStore(fakeClient, fakeClock, noObjectTTL, time.Minute)
 
 	for i := 0; i < 10; i++ {
-		store.AddReference(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i))
+		store.AddReference(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i), types.UID(fmt.Sprintf("pod-%d", i)))
 	}
 	fakeClient.ClearActions()
 
@@ -211,7 +211,7 @@ func TestCustomTTL(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Time{})
 	store := newSecretStore(fakeClient, fakeClock, customTTL, time.Minute)
 
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 	store.Get("ns", "name")
 	fakeClient.ClearActions()
 

pkg/kubelet/util/manager/manager.go

@@ -17,8 +17,9 @@ limitations under the License.
 package manager
 
 import (
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 // Manager is the interface for registering and unregistering
@@ -46,15 +47,15 @@ type Manager interface {
 // Store is the interface for a object cache that
 // can be used by cacheBasedManager.
 type Store interface {
-	// AddReference adds a reference to the object to the store.
+	// AddReference adds a reference from referencedFrom to the object to the store.
 	// Note that multiple additions to the store has to be allowed
 	// in the implementations and effectively treated as refcounted.
-	AddReference(namespace, name string)
-	// DeleteReference deletes reference to the object from the store.
+	AddReference(namespace, name string, referencedFrom types.UID)
+	// DeleteReference deletes a reference from referencedFrom to the object from the store.
 	// Note that object should be deleted only when there was a
 	// corresponding Delete call for each of Add calls (effectively
-	// when refcount was reduced to zero).
-	DeleteReference(namespace, name string)
+	// when refcount of every referenceFrom was reduced to zero).
+	DeleteReference(namespace, name string, referencedFrom types.UID)
 	// Get an object from a store.
 	Get(namespace, name string) (runtime.Object, error)
 }

pkg/kubelet/util/manager/watch_based_manager.go

@@ -21,7 +21,7 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
 	"k8s.io/klog/v2"
@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
@@ -44,7 +45,7 @@ type isImmutableFunc func(runtime.Object) bool
 
 // objectCacheItem is a single item stored in objectCache.
 type objectCacheItem struct {
-	refCount  int
+	refMap    map[types.UID]int
 	store     *cacheStore
 	reflector *cache.Reflector
 
@@ -231,7 +232,7 @@ func (c *objectCache) newReflectorLocked(namespace, name string) *objectCacheIte
 		0,
 	)
 	item := &objectCacheItem{
-		refCount:  0,
+		refMap:    make(map[types.UID]int),
 		store:     store,
 		reflector: reflector,
 		hasSynced: func() (bool, error) { return store.hasSynced(), nil },
@@ -245,7 +246,7 @@ func (c *objectCache) newReflectorLocked(namespace, name string) *objectCacheIte
 	return item
 }
 
-func (c *objectCache) AddReference(namespace, name string) {
+func (c *objectCache) AddReference(namespace, name string, referencedFrom types.UID) {
 	key := objectKey{namespace: namespace, name: name}
 
 	// AddReference is called from RegisterPod thus it needs to be efficient.
@@ -260,17 +261,20 @@ func (c *objectCache) AddReference(namespace, name string) {
 		item = c.newReflectorLocked(namespace, name)
 		c.items[key] = item
 	}
-	item.refCount++
+	item.refMap[referencedFrom]++
 }
 
-func (c *objectCache) DeleteReference(namespace, name string) {
+func (c *objectCache) DeleteReference(namespace, name string, referencedFrom types.UID) {
 	key := objectKey{namespace: namespace, name: name}
 
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	if item, ok := c.items[key]; ok {
-		item.refCount--
-		if item.refCount == 0 {
+		item.refMap[referencedFrom]--
+		if item.refMap[referencedFrom] == 0 {
+			delete(item.refMap, referencedFrom)
+		}
+		if len(item.refMap) == 0 {
 			// Stop the underlying reflector.
 			item.stop()
 			delete(c.items, key)

pkg/kubelet/util/manager/watch_based_manager_test.go

@@ -23,11 +23,12 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 
@@ -93,7 +94,7 @@ func TestSecretCache(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Now())
 	store := newSecretCache(fakeClient, fakeClock, time.Minute)
 
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 	_, err := store.Get("ns", "name")
 	if !apierrors.IsNotFound(err) {
 		t.Errorf("Expected NotFound error, got: %v", err)
@@ -138,7 +139,7 @@ func TestSecretCache(t *testing.T) {
 		t.Errorf("unexpected error: %v", err)
 	}
 
-	store.DeleteReference("ns", "name")
+	store.DeleteReference("ns", "name", "pod")
 	_, err = store.Get("ns", "name")
 	if err == nil || !strings.Contains(err.Error(), "not registered") {
 		t.Errorf("unexpected error: %v", err)
@@ -163,7 +164,7 @@ func TestSecretCacheMultipleRegistrations(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Now())
 	store := newSecretCache(fakeClient, fakeClock, time.Minute)
 
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 	// This should trigger List and Watch actions eventually.
 	actionsFn := func() (bool, error) {
 		actions := fakeClient.Actions()
@@ -184,14 +185,14 @@ func TestSecretCacheMultipleRegistrations(t *testing.T) {
 
 	// Next registrations shouldn't trigger any new actions.
 	for i := 0; i < 20; i++ {
-		store.AddReference("ns", "name")
-		store.DeleteReference("ns", "name")
+		store.AddReference("ns", "name", types.UID(fmt.Sprintf("pod-%d", i)))
+		store.DeleteReference("ns", "name", types.UID(fmt.Sprintf("pod-%d", i)))
 	}
 	actions := fakeClient.Actions()
 	assert.Equal(t, 2, len(actions), "unexpected actions: %#v", actions)
 
 	// Final delete also doesn't trigger any action.
-	store.DeleteReference("ns", "name")
+	store.DeleteReference("ns", "name", "pod")
 	_, err := store.Get("ns", "name")
 	if err == nil || !strings.Contains(err.Error(), "not registered") {
 		t.Errorf("unexpected error: %v", err)
@@ -287,7 +288,7 @@ func TestImmutableSecretStopsTheReflector(t *testing.T) {
 			}
 
 			// AddReference should start reflector.
-			store.AddReference("ns", "name")
+			store.AddReference("ns", "name", "pod")
 			if err := wait.Poll(10*time.Millisecond, time.Second, itemExists); err != nil {
 				t.Errorf("item wasn't added to cache")
 			}
@@ -375,7 +376,7 @@ func TestMaxIdleTimeStopsTheReflector(t *testing.T) {
 	}
 
 	// AddReference should start reflector.
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 	if err := wait.Poll(10*time.Millisecond, 10*time.Second, itemExists); err != nil {
 		t.Errorf("item wasn't added to cache")
 	}
@@ -467,7 +468,7 @@ func TestReflectorNotStoppedOnSlowInitialization(t *testing.T) {
 	}
 
 	// AddReference should start reflector.
-	store.AddReference("ns", "name")
+	store.AddReference("ns", "name", "pod")
 	if err := wait.Poll(10*time.Millisecond, 10*time.Second, itemExists); err != nil {
 		t.Errorf("item wasn't added to cache")
 	}
@@ -498,3 +499,123 @@ func TestReflectorNotStoppedOnSlowInitialization(t *testing.T) {
 	obj, _ := store.Get("ns", "name")
 	assert.True(t, apiequality.Semantic.DeepEqual(secret, obj))
 }
+
+func TestRefMapHandlesReferencesCorrectly(t *testing.T) {
+	secret1 := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "secret1",
+			Namespace: "ns1",
+		},
+	}
+	type step struct {
+		action         string
+		ns             string
+		name           string
+		referencedFrom types.UID
+	}
+	type expect struct {
+		ns             string
+		name           string
+		referencedFrom types.UID
+		expectCount    int
+	}
+	tests := []struct {
+		desc    string
+		steps   []step
+		expects []expect
+	}{
+		{
+			desc: "adding and deleting should works",
+			steps: []step{
+				{"add", "ns1", "secret1", "pod1"},
+				{"add", "ns1", "secret1", "pod1"},
+				{"delete", "ns1", "secret1", "pod1"},
+				{"delete", "ns1", "secret1", "pod1"},
+			},
+			expects: []expect{
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 2},
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 0},
+			},
+		},
+		{
+			desc: "deleting a non-existent reference should have no effect",
+			steps: []step{
+				{"delete", "ns1", "secret1", "pod1"},
+			},
+			expects: []expect{
+				{"ns1", "secret1", "pod1", 0},
+			},
+		},
+		{
+			desc: "deleting more than adding should not lead to negative refcount",
+			steps: []step{
+				{"add", "ns1", "secret1", "pod1"},
+				{"delete", "ns1", "secret1", "pod1"},
+				{"delete", "ns1", "secret1", "pod1"},
+			},
+			expects: []expect{
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 0},
+				{"ns1", "secret1", "pod1", 0},
+			},
+		},
+		{
+			desc: "deleting should not affect refcount of other objects or referencedFrom",
+			steps: []step{
+				{"add", "ns1", "secret1", "pod1"},
+				{"delete", "ns1", "secret1", "pod2"},
+				{"delete", "ns1", "secret2", "pod1"},
+				{"delete", "ns2", "secret1", "pod1"},
+			},
+			expects: []expect{
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 1},
+				{"ns1", "secret1", "pod1", 1},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.desc, func(t *testing.T) {
+			fakeClient := &fake.Clientset{}
+			listReactor := func(a core.Action) (bool, runtime.Object, error) {
+				result := &v1.SecretList{
+					ListMeta: metav1.ListMeta{
+						ResourceVersion: "200",
+					},
+					Items: []v1.Secret{*secret1},
+				}
+				return true, result, nil
+			}
+			fakeClient.AddReactor("list", "secrets", listReactor)
+			fakeWatch := watch.NewFake()
+			fakeClient.AddWatchReactor("secrets", core.DefaultWatchReactor(fakeWatch, nil))
+			fakeClock := testingclock.NewFakeClock(time.Now())
+			store := newSecretCache(fakeClient, fakeClock, time.Minute)
+
+			for i, step := range tc.steps {
+				expect := tc.expects[i]
+				switch step.action {
+				case "add":
+					store.AddReference(step.ns, step.name, step.referencedFrom)
+				case "delete":
+					store.DeleteReference(step.ns, step.name, step.referencedFrom)
+				default:
+					t.Errorf("unrecognized action of testcase %v", tc.desc)
+				}
+
+				key := objectKey{namespace: expect.ns, name: expect.name}
+				item, exists := store.items[key]
+				if !exists {
+					if tc.expects[i].expectCount != 0 {
+						t.Errorf("reference to %v/%v from %v should exists", expect.ns, expect.name, expect.referencedFrom)
+					}
+				} else if item.refMap[expect.referencedFrom] != expect.expectCount {
+					t.Errorf("expects %v but got %v", expect.expectCount, item.refMap[expect.referencedFrom])
+				}
+			}
+		})
+	}
+}

test/integration/kubelet/watch_manager_test.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
@@ -107,7 +108,7 @@ func TestWatchBasedManager(t *testing.T) {
 			for j := 0; j < 100; j++ {
 				name := fmt.Sprintf("s%d", i*100+j)
 				start := time.Now()
-				store.AddReference(testNamespace, name)
+				store.AddReference(testNamespace, name, types.UID(name))
 				err := wait.PollImmediate(10*time.Millisecond, 10*time.Second, func() (bool, error) {
 					obj, err := store.Get(testNamespace, name)
 					if err != nil {