diff --git a/pkg/kubelet/configmap/BUILD b/pkg/kubelet/configmap/BUILD index 4762f1e9b2c..84d1e029d7f 100644 --- a/pkg/kubelet/configmap/BUILD +++ b/pkg/kubelet/configmap/BUILD @@ -15,13 +15,12 @@ go_library( importpath = "k8s.io/kubernetes/pkg/kubelet/configmap", deps = [ "//pkg/api/v1/pod:go_default_library", - "//pkg/kubelet/util:go_default_library", + "//pkg/kubelet/util/manager:go_default_library", "//vendor/k8s.io/api/core/v1:go_default_library", - "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library", "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library", "//vendor/k8s.io/apimachinery/pkg/util/clock:go_default_library", "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library", - "//vendor/k8s.io/apiserver/pkg/storage/etcd:go_default_library", "//vendor/k8s.io/client-go/kubernetes:go_default_library", ], ) @@ -44,13 +43,12 @@ go_test( srcs = ["configmap_manager_test.go"], embed = [":go_default_library"], deps = [ - "//vendor/github.com/stretchr/testify/assert:go_default_library", + "//pkg/kubelet/util/manager:go_default_library", "//vendor/k8s.io/api/core/v1:go_default_library", - "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library", "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library", "//vendor/k8s.io/apimachinery/pkg/util/clock:go_default_library", + "//vendor/k8s.io/client-go/kubernetes:go_default_library", "//vendor/k8s.io/client-go/kubernetes/fake:go_default_library", - "//vendor/k8s.io/client-go/testing:go_default_library", ], ) diff --git a/pkg/kubelet/configmap/configmap_manager.go b/pkg/kubelet/configmap/configmap_manager.go index 71a7ae14854..91c75ddca72 100644 --- a/pkg/kubelet/configmap/configmap_manager.go +++ b/pkg/kubelet/configmap/configmap_manager.go @@ -18,26 +18,19 @@ package configmap import ( "fmt" - "strconv" - "sync" "time" "k8s.io/api/core/v1" - storageetcd "k8s.io/apiserver/pkg/storage/etcd" clientset "k8s.io/client-go/kubernetes" podutil "k8s.io/kubernetes/pkg/api/v1/pod" - "k8s.io/kubernetes/pkg/kubelet/util" + "k8s.io/kubernetes/pkg/kubelet/util/manager" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/clock" "k8s.io/apimachinery/pkg/util/sets" ) -const ( - defaultTTL = time.Minute -) - type Manager interface { // Get configmap by configmap namespace and name. GetConfigMap(namespace, name string) (*v1.ConfigMap, error) @@ -73,191 +66,31 @@ func (s *simpleConfigMapManager) RegisterPod(pod *v1.Pod) { func (s *simpleConfigMapManager) UnregisterPod(pod *v1.Pod) { } -type GetObjectTTLFunc func() (time.Duration, bool) - -type objectKey struct { - namespace string - name string +// configMapManager keeps a cache of all configmaps necessary +// for registered pods. Different implementation of the store +// may result in different semantics for freshness of configmaps +// (e.g. ttl-based implementation vs watch-based implementation). +type configMapManager struct { + manager manager.Manager } -// configMapStoreItems is a single item stored in configMapStore. -type configMapStoreItem struct { - refCount int - configMap *configMapData -} - -type configMapData struct { - sync.Mutex - - configMap *v1.ConfigMap - err error - lastUpdateTime time.Time -} - -// configMapStore is a local cache of configmaps. -type configMapStore struct { - kubeClient clientset.Interface - clock clock.Clock - - lock sync.Mutex - items map[objectKey]*configMapStoreItem - - defaultTTL time.Duration - getTTL GetObjectTTLFunc -} - -func newConfigMapStore(kubeClient clientset.Interface, clock clock.Clock, getTTL GetObjectTTLFunc, ttl time.Duration) *configMapStore { - return &configMapStore{ - kubeClient: kubeClient, - clock: clock, - items: make(map[objectKey]*configMapStoreItem), - defaultTTL: ttl, - getTTL: getTTL, +func (c *configMapManager) GetConfigMap(namespace, name string) (*v1.ConfigMap, error) { + object, err := c.manager.GetObject(namespace, name) + if err != nil { + return nil, err } -} - -func isConfigMapOlder(newConfigMap, oldConfigMap *v1.ConfigMap) bool { - if newConfigMap == nil || oldConfigMap == nil { - return false + if configmap, ok := object.(*v1.ConfigMap); ok { + return configmap, nil } - newVersion, _ := storageetcd.Versioner.ObjectResourceVersion(newConfigMap) - oldVersion, _ := storageetcd.Versioner.ObjectResourceVersion(oldConfigMap) - return newVersion < oldVersion + return nil, fmt.Errorf("unexpected object type: %v", object) } -func (s *configMapStore) Add(namespace, name string) { - key := objectKey{namespace: namespace, name: name} - - // Add is called from RegisterPod, thus it needs to be efficient. - // Thus Add() is only increasing refCount and generation of a given configmap. - // Then Get() is responsible for fetching if needed. - s.lock.Lock() - defer s.lock.Unlock() - item, exists := s.items[key] - if !exists { - item = &configMapStoreItem{ - refCount: 0, - configMap: &configMapData{}, - } - s.items[key] = item - } - - item.refCount++ - // This will trigger fetch on the next Get() operation. - item.configMap = nil +func (c *configMapManager) RegisterPod(pod *v1.Pod) { + c.manager.RegisterPod(pod) } -func (s *configMapStore) Delete(namespace, name string) { - key := objectKey{namespace: namespace, name: name} - - s.lock.Lock() - defer s.lock.Unlock() - if item, ok := s.items[key]; ok { - item.refCount-- - if item.refCount == 0 { - delete(s.items, key) - } - } -} - -func GetObjectTTLFromNodeFunc(getNode func() (*v1.Node, error)) GetObjectTTLFunc { - return func() (time.Duration, bool) { - node, err := getNode() - if err != nil { - return time.Duration(0), false - } - if node != nil && node.Annotations != nil { - if value, ok := node.Annotations[v1.ObjectTTLAnnotationKey]; ok { - if intValue, err := strconv.Atoi(value); err == nil { - return time.Duration(intValue) * time.Second, true - } - } - } - return time.Duration(0), false - } -} - -func (s *configMapStore) isConfigMapFresh(data *configMapData) bool { - configMapTTL := s.defaultTTL - if ttl, ok := s.getTTL(); ok { - configMapTTL = ttl - } - return s.clock.Now().Before(data.lastUpdateTime.Add(configMapTTL)) -} - -func (s *configMapStore) Get(namespace, name string) (*v1.ConfigMap, error) { - key := objectKey{namespace: namespace, name: name} - - data := func() *configMapData { - s.lock.Lock() - defer s.lock.Unlock() - item, exists := s.items[key] - if !exists { - return nil - } - if item.configMap == nil { - item.configMap = &configMapData{} - } - return item.configMap - }() - if data == nil { - return nil, fmt.Errorf("configmap %q/%q not registered", namespace, name) - } - - // After updating data in configMapStore, lock the data, fetch configMap if - // needed and return data. - data.Lock() - defer data.Unlock() - if data.err != nil || !s.isConfigMapFresh(data) { - opts := metav1.GetOptions{} - if data.configMap != nil && data.err == nil { - // This is just a periodic refresh of a configmap we successfully fetched previously. - // In this case, server data from apiserver cache to reduce the load on both - // etcd and apiserver (the cache is eventually consistent). - util.FromApiserverCache(&opts) - } - configMap, err := s.kubeClient.CoreV1().ConfigMaps(namespace).Get(name, opts) - if err != nil && !apierrors.IsNotFound(err) && data.configMap == nil && data.err == nil { - // Couldn't fetch the latest configmap, but there is no cached data to return. - // Return the fetch result instead. - return configMap, err - } - if (err == nil && !isConfigMapOlder(configMap, data.configMap)) || apierrors.IsNotFound(err) { - // If the fetch succeeded with a newer version of the configmap, or if the - // configmap could not be found in the apiserver, update the cached data to - // reflect the current status. - data.configMap = configMap - data.err = err - data.lastUpdateTime = s.clock.Now() - } - } - return data.configMap, data.err -} - -// cachingConfigMapManager keeps a cache of all configmaps necessary for registered pods. -// It implements the following logic: -// - whenever a pod is created or updated, the cached versions of all its configmaps -// are invalidated -// - every GetConfigMap() call tries to fetch the value from local cache; if it is -// not there, invalidated or too old, we fetch it from apiserver and refresh the -// value in cache; otherwise it is just fetched from cache -type cachingConfigMapManager struct { - configMapStore *configMapStore - - lock sync.Mutex - registeredPods map[objectKey]*v1.Pod -} - -func NewCachingConfigMapManager(kubeClient clientset.Interface, getTTL GetObjectTTLFunc) Manager { - csm := &cachingConfigMapManager{ - configMapStore: newConfigMapStore(kubeClient, clock.RealClock{}, getTTL, defaultTTL), - registeredPods: make(map[objectKey]*v1.Pod), - } - return csm -} - -func (c *cachingConfigMapManager) GetConfigMap(namespace, name string) (*v1.ConfigMap, error) { - return c.configMapStore.Get(namespace, name) +func (c *configMapManager) UnregisterPod(pod *v1.Pod) { + c.manager.UnregisterPod(pod) } func getConfigMapNames(pod *v1.Pod) sets.String { @@ -269,39 +102,24 @@ func getConfigMapNames(pod *v1.Pod) sets.String { return result } -func (c *cachingConfigMapManager) RegisterPod(pod *v1.Pod) { - names := getConfigMapNames(pod) - c.lock.Lock() - defer c.lock.Unlock() - for name := range names { - c.configMapStore.Add(pod.Namespace, name) - } - var prev *v1.Pod - key := objectKey{namespace: pod.Namespace, name: pod.Name} - prev = c.registeredPods[key] - c.registeredPods[key] = pod - if prev != nil { - for name := range getConfigMapNames(prev) { - // On an update, the .Add() call above will have re-incremented the - // ref count of any existing items, so any configmaps that are in both - // 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.configMapStore.Delete(prev.Namespace, name) - } - } -} +const ( + defaultTTL = time.Minute +) -func (c *cachingConfigMapManager) UnregisterPod(pod *v1.Pod) { - var prev *v1.Pod - key := objectKey{namespace: pod.Namespace, name: pod.Name} - c.lock.Lock() - defer c.lock.Unlock() - prev = c.registeredPods[key] - delete(c.registeredPods, key) - if prev != nil { - for name := range getConfigMapNames(prev) { - c.configMapStore.Delete(prev.Namespace, name) - } +// NewCachingConfigMapManager creates a manager that keeps a cache of all configmaps +// necessary for registered pods. +// It implement the following logic: +// - whenever a pod is create or updated, the cached versions of all configmaps +// are invalidated +// - every GetObject() call tries to fetch the value from local cache; if it is +// not there, invalidated or too old, we fetch it from apiserver and refresh the +// value in cache; otherwise it is just fetched from cache +func NewCachingConfigMapManager(kubeClient clientset.Interface, getTTL manager.GetObjectTTLFunc) Manager { + getConfigMap := func(namespace, name string, opts metav1.GetOptions) (runtime.Object, error) { + return kubeClient.CoreV1().ConfigMaps(namespace).Get(name, opts) + } + configMapStore := manager.NewObjectStore(getConfigMap, clock.RealClock{}, getTTL, defaultTTL) + return &configMapManager{ + manager: manager.NewCacheBasedManager(configMapStore, getConfigMapNames), } } diff --git a/pkg/kubelet/configmap/configmap_manager_test.go b/pkg/kubelet/configmap/configmap_manager_test.go index d79a80b5012..2b8433c0e43 100644 --- a/pkg/kubelet/configmap/configmap_manager_test.go +++ b/pkg/kubelet/configmap/configmap_manager_test.go @@ -18,30 +18,27 @@ package configmap import ( "fmt" - "reflect" "strings" - "sync" "testing" "time" "k8s.io/api/core/v1" - "k8s.io/client-go/kubernetes/fake" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/clock" - core "k8s.io/client-go/testing" - "github.com/stretchr/testify/assert" + clientset "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/kubernetes/pkg/kubelet/util/manager" ) -func checkConfigMap(t *testing.T, store *configMapStore, ns, name string, shouldExist bool) { +func checkObject(t *testing.T, store manager.Store, ns, name string, shouldExist bool) { _, err := store.Get(ns, name) if shouldExist && err != nil { t.Errorf("unexpected actions: %#v", err) } - if !shouldExist && (err == nil || !strings.Contains(err.Error(), fmt.Sprintf("configmap %q/%q not registered", ns, name))) { + if !shouldExist && (err == nil || !strings.Contains(err.Error(), fmt.Sprintf("object %q/%q not registered", ns, name))) { t.Errorf("unexpected actions: %#v", err) } } @@ -50,242 +47,9 @@ func noObjectTTL() (time.Duration, bool) { return time.Duration(0), false } -func TestConfigMapStore(t *testing.T) { - fakeClient := &fake.Clientset{} - store := newConfigMapStore(fakeClient, clock.RealClock{}, noObjectTTL, 0) - store.Add("ns1", "name1") - store.Add("ns2", "name2") - store.Add("ns1", "name1") - store.Add("ns1", "name1") - store.Delete("ns1", "name1") - store.Delete("ns2", "name2") - store.Add("ns3", "name3") - - // Adds don't issue Get requests. - actions := fakeClient.Actions() - assert.Equal(t, 0, len(actions), "unexpected actions: %#v", actions) - // Should issue Get request - store.Get("ns1", "name1") - // Shouldn't issue Get request, as configMap is not registered - store.Get("ns2", "name2") - // Should issue Get request - store.Get("ns3", "name3") - - actions = fakeClient.Actions() - assert.Equal(t, 2, len(actions), "unexpected actions: %#v", actions) - - for _, a := range actions { - assert.True(t, a.Matches("get", "configmaps"), "unexpected actions: %#v", a) - } - - checkConfigMap(t, store, "ns1", "name1", true) - checkConfigMap(t, store, "ns2", "name2", false) - checkConfigMap(t, store, "ns3", "name3", true) - checkConfigMap(t, store, "ns4", "name4", false) -} - -func TestConfigMapStoreDeletingConfigMap(t *testing.T) { - fakeClient := &fake.Clientset{} - store := newConfigMapStore(fakeClient, clock.RealClock{}, noObjectTTL, 0) - store.Add("ns", "name") - - result := &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name", ResourceVersion: "10"}} - fakeClient.AddReactor("get", "configmaps", func(action core.Action) (bool, runtime.Object, error) { - return true, result, nil - }) - configMap, err := store.Get("ns", "name") - if err != nil { - t.Errorf("Unexpected error: %v", err) - } - if !reflect.DeepEqual(configMap, result) { - t.Errorf("Unexpected configMap: %v", configMap) - } - - fakeClient.PrependReactor("get", "configmaps", func(action core.Action) (bool, runtime.Object, error) { - return true, &v1.ConfigMap{}, apierrors.NewNotFound(v1.Resource("configMap"), "name") - }) - configMap, err = store.Get("ns", "name") - if err == nil || !apierrors.IsNotFound(err) { - t.Errorf("Unexpected error: %v", err) - } - if !reflect.DeepEqual(configMap, &v1.ConfigMap{}) { - t.Errorf("Unexpected configMap: %v", configMap) - } -} - -func TestConfigMapStoreGetAlwaysRefresh(t *testing.T) { - fakeClient := &fake.Clientset{} - fakeClock := clock.NewFakeClock(time.Now()) - store := newConfigMapStore(fakeClient, fakeClock, noObjectTTL, 0) - - for i := 0; i < 10; i++ { - store.Add(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i)) - } - fakeClient.ClearActions() - - wg := sync.WaitGroup{} - wg.Add(100) - for i := 0; i < 100; i++ { - go func(i int) { - store.Get(fmt.Sprintf("ns-%d", i%10), fmt.Sprintf("name-%d", i%10)) - wg.Done() - }(i) - } - wg.Wait() - actions := fakeClient.Actions() - assert.Equal(t, 100, len(actions), "unexpected actions: %#v", actions) - - for _, a := range actions { - assert.True(t, a.Matches("get", "configmaps"), "unexpected actions: %#v", a) - } -} - -func TestConfigMapStoreGetNeverRefresh(t *testing.T) { - fakeClient := &fake.Clientset{} - fakeClock := clock.NewFakeClock(time.Now()) - store := newConfigMapStore(fakeClient, fakeClock, noObjectTTL, time.Minute) - - for i := 0; i < 10; i++ { - store.Add(fmt.Sprintf("ns-%d", i), fmt.Sprintf("name-%d", i)) - } - fakeClient.ClearActions() - - wg := sync.WaitGroup{} - wg.Add(100) - for i := 0; i < 100; i++ { - go func(i int) { - store.Get(fmt.Sprintf("ns-%d", i%10), fmt.Sprintf("name-%d", i%10)) - wg.Done() - }(i) - } - wg.Wait() - actions := fakeClient.Actions() - // Only first Get, should forward the Get request. - assert.Equal(t, 10, len(actions), "unexpected actions: %#v", actions) -} - -func TestCustomTTL(t *testing.T) { - ttl := time.Duration(0) - ttlExists := false - customTTL := func() (time.Duration, bool) { - return ttl, ttlExists - } - - fakeClient := &fake.Clientset{} - fakeClock := clock.NewFakeClock(time.Time{}) - store := newConfigMapStore(fakeClient, fakeClock, customTTL, time.Minute) - - store.Add("ns", "name") - store.Get("ns", "name") - fakeClient.ClearActions() - - // Set 0-ttl and see if that works. - ttl = time.Duration(0) - ttlExists = true - store.Get("ns", "name") - actions := fakeClient.Actions() - assert.Equal(t, 1, len(actions), "unexpected actions: %#v", actions) - fakeClient.ClearActions() - - // Set 5-minute ttl and see if this works. - ttl = time.Duration(5) * time.Minute - store.Get("ns", "name") - actions = fakeClient.Actions() - assert.Equal(t, 0, len(actions), "unexpected actions: %#v", actions) - // Still no effect after 4 minutes. - fakeClock.Step(4 * time.Minute) - store.Get("ns", "name") - actions = fakeClient.Actions() - assert.Equal(t, 0, len(actions), "unexpected actions: %#v", actions) - // Now it should have an effect. - fakeClock.Step(time.Minute) - store.Get("ns", "name") - actions = fakeClient.Actions() - assert.Equal(t, 1, len(actions), "unexpected actions: %#v", actions) - fakeClient.ClearActions() - - // Now remove the custom ttl and see if that works. - ttlExists = false - fakeClock.Step(55 * time.Second) - store.Get("ns", "name") - actions = fakeClient.Actions() - assert.Equal(t, 0, len(actions), "unexpected actions: %#v", actions) - // Pass the minute and it should be triggered now. - fakeClock.Step(5 * time.Second) - store.Get("ns", "name") - actions = fakeClient.Actions() - assert.Equal(t, 1, len(actions), "unexpected actions: %#v", actions) -} - -func TestParseNodeAnnotation(t *testing.T) { - testCases := []struct { - node *v1.Node - err error - exists bool - ttl time.Duration - }{ - { - node: nil, - err: fmt.Errorf("error"), - exists: false, - }, - { - node: &v1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "node", - }, - }, - exists: false, - }, - { - node: &v1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "node", - Annotations: map[string]string{}, - }, - }, - exists: false, - }, - { - node: &v1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "node", - Annotations: map[string]string{v1.ObjectTTLAnnotationKey: "bad"}, - }, - }, - exists: false, - }, - { - node: &v1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "node", - Annotations: map[string]string{v1.ObjectTTLAnnotationKey: "0"}, - }, - }, - exists: true, - ttl: time.Duration(0), - }, - { - node: &v1.Node{ - ObjectMeta: metav1.ObjectMeta{ - Name: "node", - Annotations: map[string]string{v1.ObjectTTLAnnotationKey: "60"}, - }, - }, - exists: true, - ttl: time.Minute, - }, - } - for i, testCase := range testCases { - getNode := func() (*v1.Node, error) { return testCase.node, testCase.err } - ttl, exists := GetObjectTTLFromNodeFunc(getNode)() - if exists != testCase.exists { - t.Errorf("%d: incorrect parsing: %t", i, exists) - continue - } - if exists && ttl != testCase.ttl { - t.Errorf("%d: incorrect ttl: %v", i, ttl) - } +func getConfigMap(fakeClient clientset.Interface) manager.GetObjectFunc { + return func(namespace, name string, opts metav1.GetOptions) (runtime.Object, error) { + return fakeClient.CoreV1().ConfigMaps(namespace).Get(name, opts) } } @@ -348,152 +112,11 @@ func podWithConfigMaps(ns, podName string, toAttach configMapsToAttach) *v1.Pod return pod } -func TestCacheInvalidation(t *testing.T) { +func TestCacheBasedConfigMapManager(t *testing.T) { fakeClient := &fake.Clientset{} - fakeClock := clock.NewFakeClock(time.Now()) - store := newConfigMapStore(fakeClient, fakeClock, noObjectTTL, time.Minute) - manager := &cachingConfigMapManager{ - configMapStore: store, - registeredPods: make(map[objectKey]*v1.Pod), - } - - // Create a pod with some configMaps. - s1 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s1"}, envFromNames: []string{"s10"}}, - {envVarNames: []string{"s2"}}, - }, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name1", s1)) - // Fetch both configMaps - this should triggger get operations. - store.Get("ns1", "s1") - store.Get("ns1", "s10") - store.Get("ns1", "s2") - actions := fakeClient.Actions() - assert.Equal(t, 3, len(actions), "unexpected actions: %#v", actions) - fakeClient.ClearActions() - - // Update a pod with a new configMap. - s2 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s1"}}, - {envVarNames: []string{"s2"}, envFromNames: []string{"s20"}}, - }, - volumes: []string{"s3"}, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name1", s2)) - // All configMaps should be invalidated - this should trigger get operations. - store.Get("ns1", "s1") - store.Get("ns1", "s2") - store.Get("ns1", "s20") - store.Get("ns1", "s3") - actions = fakeClient.Actions() - assert.Equal(t, 4, len(actions), "unexpected actions: %#v", actions) - fakeClient.ClearActions() - - // Create a new pod that is refencing the first three configMaps - those should - // be invalidated. - manager.RegisterPod(podWithConfigMaps("ns1", "name2", s1)) - store.Get("ns1", "s1") - store.Get("ns1", "s10") - store.Get("ns1", "s2") - store.Get("ns1", "s20") - store.Get("ns1", "s3") - actions = fakeClient.Actions() - assert.Equal(t, 3, len(actions), "unexpected actions: %#v", actions) - fakeClient.ClearActions() -} - -func TestCacheRefcounts(t *testing.T) { - fakeClient := &fake.Clientset{} - fakeClock := clock.NewFakeClock(time.Now()) - store := newConfigMapStore(fakeClient, fakeClock, noObjectTTL, time.Minute) - manager := &cachingConfigMapManager{ - configMapStore: store, - registeredPods: make(map[objectKey]*v1.Pod), - } - - s1 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s1"}, envFromNames: []string{"s10"}}, - {envVarNames: []string{"s2"}}, - }, - volumes: []string{"s3"}, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name1", s1)) - manager.RegisterPod(podWithConfigMaps("ns1", "name2", s1)) - s2 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s4"}}, - {envVarNames: []string{"s5"}, envFromNames: []string{"s50"}}, - }, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name2", s2)) - manager.RegisterPod(podWithConfigMaps("ns1", "name3", s2)) - manager.RegisterPod(podWithConfigMaps("ns1", "name4", s2)) - manager.UnregisterPod(podWithConfigMaps("ns1", "name3", s2)) - s3 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s3"}, envFromNames: []string{"s30"}}, - {envVarNames: []string{"s5"}}, - }, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name5", s3)) - manager.RegisterPod(podWithConfigMaps("ns1", "name6", s3)) - s4 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s6"}}, - {envFromNames: []string{"s60"}}, - }, - } - manager.RegisterPod(podWithConfigMaps("ns1", "name7", s4)) - manager.UnregisterPod(podWithConfigMaps("ns1", "name7", s4)) - - // Also check the Add + Update + Remove scenario. - manager.RegisterPod(podWithConfigMaps("ns1", "other-name", s1)) - manager.RegisterPod(podWithConfigMaps("ns1", "other-name", s2)) - manager.UnregisterPod(podWithConfigMaps("ns1", "other-name", s2)) - - s5 := configMapsToAttach{ - containerEnvConfigMaps: []envConfigMaps{ - {envVarNames: []string{"s7"}}, - {envFromNames: []string{"s70"}}, - }, - } - - // Check the no-op update scenario - manager.RegisterPod(podWithConfigMaps("ns1", "noop-pod", s5)) - manager.RegisterPod(podWithConfigMaps("ns1", "noop-pod", s5)) - - refs := func(ns, name string) int { - store.lock.Lock() - defer store.lock.Unlock() - item, ok := store.items[objectKey{ns, name}] - if !ok { - return 0 - } - return item.refCount - } - assert.Equal(t, 1, refs("ns1", "s1")) - assert.Equal(t, 1, refs("ns1", "s10")) - assert.Equal(t, 1, refs("ns1", "s2")) - assert.Equal(t, 3, refs("ns1", "s3")) - assert.Equal(t, 2, refs("ns1", "s30")) - assert.Equal(t, 2, refs("ns1", "s4")) - assert.Equal(t, 4, refs("ns1", "s5")) - assert.Equal(t, 2, refs("ns1", "s50")) - assert.Equal(t, 0, refs("ns1", "s6")) - assert.Equal(t, 0, refs("ns1", "s60")) - assert.Equal(t, 1, refs("ns1", "s7")) - assert.Equal(t, 1, refs("ns1", "s70")) -} - -func TestCachingConfigMapManager(t *testing.T) { - fakeClient := &fake.Clientset{} - configMapStore := newConfigMapStore(fakeClient, clock.RealClock{}, noObjectTTL, 0) - manager := &cachingConfigMapManager{ - configMapStore: configMapStore, - registeredPods: make(map[objectKey]*v1.Pod), + store := manager.NewObjectStore(getConfigMap(fakeClient), clock.RealClock{}, noObjectTTL, 0) + manager := &configMapManager{ + manager: manager.NewCacheBasedManager(store, getConfigMapNames), } // Create a pod with some configMaps. @@ -543,7 +166,7 @@ func TestCachingConfigMapManager(t *testing.T) { for _, ns := range []string{"ns1", "ns2", "ns3"} { for _, configMap := range []string{"s1", "s2", "s3", "s4", "s5", "s6", "s20", "s40", "s50"} { - checkConfigMap(t, configMapStore, ns, configMap, shouldExist(ns, configMap)) + checkObject(t, store, ns, configMap, shouldExist(ns, configMap)) } } } diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index fb0f5321726..e3d2df76fed 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -547,7 +547,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration, klet.secretManager = secretManager configMapManager := configmap.NewCachingConfigMapManager( - kubeDeps.KubeClient, configmap.GetObjectTTLFromNodeFunc(klet.GetNode)) + kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode)) klet.configMapManager = configMapManager if klet.experimentalHostUserNamespaceDefaulting { diff --git a/pkg/kubelet/secret/secret_manager.go b/pkg/kubelet/secret/secret_manager.go index d59260447fd..a339694138b 100644 --- a/pkg/kubelet/secret/secret_manager.go +++ b/pkg/kubelet/secret/secret_manager.go @@ -66,16 +66,16 @@ func (s *simpleSecretManager) RegisterPod(pod *v1.Pod) { func (s *simpleSecretManager) UnregisterPod(pod *v1.Pod) { } -// cachingSecretManager keeps a store with secrets necessary +// secretManager keeps a store with secrets necessary // for registered pods. Different implementations of the store // may result in different semantics for freshness of secrets // (e.g. ttl-based implementation vs watch-based implementation). -type cachingSecretManager struct { +type secretManager struct { manager manager.Manager } -func (c *cachingSecretManager) GetSecret(namespace, name string) (*v1.Secret, error) { - object, err := c.manager.GetObject(namespace, name) +func (s *secretManager) GetSecret(namespace, name string) (*v1.Secret, error) { + object, err := s.manager.GetObject(namespace, name) if err != nil { return nil, err } @@ -85,12 +85,12 @@ func (c *cachingSecretManager) GetSecret(namespace, name string) (*v1.Secret, er return nil, fmt.Errorf("unexpected object type: %v", object) } -func (c *cachingSecretManager) RegisterPod(pod *v1.Pod) { - c.manager.RegisterPod(pod) +func (s *secretManager) RegisterPod(pod *v1.Pod) { + s.manager.RegisterPod(pod) } -func (c *cachingSecretManager) UnregisterPod(pod *v1.Pod) { - c.manager.UnregisterPod(pod) +func (s *secretManager) UnregisterPod(pod *v1.Pod) { + s.manager.UnregisterPod(pod) } func getSecretNames(pod *v1.Pod) sets.String { @@ -106,7 +106,7 @@ const ( defaultTTL = time.Minute ) -// NewCacheBasedManager creates a manager that keeps a cache of all secrets +// NewCachingSecretManager creates a manager that keeps a cache of all secrets // necessary for registered pods. // It implements the following logic: // - whenever a pod is created or updated, the cached versions of all secrets @@ -119,7 +119,7 @@ func NewCachingSecretManager(kubeClient clientset.Interface, getTTL manager.GetO return kubeClient.CoreV1().Secrets(namespace).Get(name, opts) } secretStore := manager.NewObjectStore(getSecret, clock.RealClock{}, getTTL, defaultTTL) - return &cachingSecretManager{ + return &secretManager{ manager: manager.NewCacheBasedManager(secretStore, getSecretNames), } } diff --git a/pkg/kubelet/secret/secret_manager_test.go b/pkg/kubelet/secret/secret_manager_test.go index 4295246935f..0dba94a0262 100644 --- a/pkg/kubelet/secret/secret_manager_test.go +++ b/pkg/kubelet/secret/secret_manager_test.go @@ -108,7 +108,7 @@ func podWithSecrets(ns, podName string, toAttach secretsToAttach) *v1.Pod { func TestCacheBasedSecretManager(t *testing.T) { fakeClient := &fake.Clientset{} store := manager.NewObjectStore(getSecret(fakeClient), clock.RealClock{}, noObjectTTL, 0) - manager := &cachingSecretManager{ + manager := &secretManager{ manager: manager.NewCacheBasedManager(store, getSecretNames), }