Enable watching secret and configmap manager

pkg/kubelet/apis/kubeletconfig/v1beta1/zz_generated.conversion.go

@@ -253,6 +253,7 @@ func autoConvert_v1beta1_KubeletConfiguration_To_kubeletconfig_KubeletConfigurat
 	if err := v1.Convert_Pointer_int32_To_int32(&in.ContainerLogMaxFiles, &out.ContainerLogMaxFiles, s); err != nil {
 		return err
 	}
+	out.ConfigMapAndSecretChangeDetectionStrategy = kubeletconfig.ResourceChangeDetectionStrategy(in.ConfigMapAndSecretChangeDetectionStrategy)
 	out.SystemReserved = *(*map[string]string)(unsafe.Pointer(&in.SystemReserved))
 	out.KubeReserved = *(*map[string]string)(unsafe.Pointer(&in.KubeReserved))
 	out.SystemReservedCgroup = in.SystemReservedCgroup
@@ -377,6 +378,7 @@ func autoConvert_kubeletconfig_KubeletConfiguration_To_v1beta1_KubeletConfigurat
 	if err := v1.Convert_int32_To_Pointer_int32(&in.ContainerLogMaxFiles, &out.ContainerLogMaxFiles, s); err != nil {
 		return err
 	}
+	out.ConfigMapAndSecretChangeDetectionStrategy = ResourceChangeDetectionStrategy(in.ConfigMapAndSecretChangeDetectionStrategy)
 	out.SystemReserved = *(*map[string]string)(unsafe.Pointer(&in.SystemReserved))
 	out.KubeReserved = *(*map[string]string)(unsafe.Pointer(&in.KubeReserved))
 	out.SystemReservedCgroup = in.SystemReservedCgroup

pkg/kubelet/configmap/BUILD

@@ -15,12 +15,14 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/kubelet/configmap",
     deps = [
         "//pkg/api/v1/pod:go_default_library",
+        "//pkg/apis/core/v1:go_default_library",
         "//pkg/kubelet/util/manager:go_default_library",
         "//vendor/k8s.io/api/core/v1: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/apimachinery/pkg/watch:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
     ],
 )

pkg/kubelet/configmap/configmap_manager.go

@@ -23,12 +23,14 @@ import (
 	"k8s.io/api/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
+	corev1 "k8s.io/kubernetes/pkg/apis/core/v1"
 	"k8s.io/kubernetes/pkg/kubelet/util/manager"
 
 	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"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 type Manager interface {
@@ -123,3 +125,25 @@ func NewCachingConfigMapManager(kubeClient clientset.Interface, getTTL manager.G
 		manager: manager.NewCacheBasedManager(configMapStore, getConfigMapNames),
 	}
 }
+
+// NewWatchingConfigMapManager creates a manager that keeps a cache of all configmaps
+// necessary for registered pods.
+// It implements the following logic:
+// - whenever a pod is created or updated, we start inidvidual watches for all
+//   referenced objects that aren't referenced from other registered pods
+// - every GetObject() returns a value from local cache propagated via watches
+func NewWatchingConfigMapManager(kubeClient clientset.Interface) Manager {
+	listConfigMap := func(namespace string, opts metav1.ListOptions) (runtime.Object, error) {
+		return kubeClient.CoreV1().ConfigMaps(namespace).List(opts)
+	}
+	watchConfigMap := func(namespace string, opts metav1.ListOptions) (watch.Interface, error) {
+		return kubeClient.CoreV1().ConfigMaps(namespace).Watch(opts)
+	}
+	newConfigMap := func() runtime.Object {
+		return &v1.ConfigMap{}
+	}
+	gr := corev1.Resource("configmap")
+	return &configMapManager{
+		manager: manager.NewWatchBasedManager(listConfigMap, watchConfigMap, newConfigMap, gr, getConfigMapNames),
+	}
+}

pkg/kubelet/kubelet.go

@@ -551,12 +551,25 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		klet.cloudproviderRequestTimeout = 10 * time.Second
 	}
 
-	secretManager := secret.NewCachingSecretManager(
-		kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
-	klet.secretManager = secretManager
+	var secretManager secret.Manager
+	var configMapManager configmap.Manager
+	switch kubeCfg.ConfigMapAndSecretChangeDetectionStrategy {
+	case kubeletconfiginternal.WatchChangeDetectionStrategy:
+		secretManager = secret.NewWatchingSecretManager(kubeDeps.KubeClient)
+		configMapManager = configmap.NewWatchingConfigMapManager(kubeDeps.KubeClient)
+	case kubeletconfiginternal.TTLCacheChangeDetectionStrategy:
+		secretManager = secret.NewCachingSecretManager(
+			kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
+		configMapManager = configmap.NewCachingConfigMapManager(
+			kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
+	case kubeletconfiginternal.GetChangeDetectionStrategy:
+		secretManager = secret.NewSimpleSecretManager(kubeDeps.KubeClient)
+		configMapManager = configmap.NewSimpleConfigMapManager(kubeDeps.KubeClient)
+	default:
+		return nil, fmt.Errorf("unknown configmap and secret manager mode: %v", kubeCfg.ConfigMapAndSecretChangeDetectionStrategy)
+	}
 
-	configMapManager := configmap.NewCachingConfigMapManager(
-		kubeDeps.KubeClient, manager.GetObjectTTLFromNodeFunc(klet.GetNode))
+	klet.secretManager = secretManager
 	klet.configMapManager = configMapManager
 
 	if klet.experimentalHostUserNamespaceDefaulting {

pkg/kubelet/secret/BUILD

@@ -30,12 +30,14 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/kubelet/secret",
     deps = [
         "//pkg/api/v1/pod:go_default_library",
+        "//pkg/apis/core/v1:go_default_library",
         "//pkg/kubelet/util/manager:go_default_library",
         "//vendor/k8s.io/api/core/v1: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/apimachinery/pkg/watch:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
     ],
 )

pkg/kubelet/secret/secret_manager.go

@@ -23,12 +23,14 @@ import (
 	"k8s.io/api/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
+	corev1 "k8s.io/kubernetes/pkg/apis/core/v1"
 	"k8s.io/kubernetes/pkg/kubelet/util/manager"
 
 	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"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 type Manager interface {
@@ -123,3 +125,25 @@ func NewCachingSecretManager(kubeClient clientset.Interface, getTTL manager.GetO
 		manager: manager.NewCacheBasedManager(secretStore, getSecretNames),
 	}
 }
+
+// NewWatchingSecretManager 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, we start inidvidual watches for all
+//   referenced objects that aren't referenced from other registered pods
+// - every GetObject() returns a value from local cache propagated via watches
+func NewWatchingSecretManager(kubeClient clientset.Interface) Manager {
+	listSecret := func(namespace string, opts metav1.ListOptions) (runtime.Object, error) {
+		return kubeClient.CoreV1().Secrets(namespace).List(opts)
+	}
+	watchSecret := func(namespace string, opts metav1.ListOptions) (watch.Interface, error) {
+		return kubeClient.CoreV1().Secrets(namespace).Watch(opts)
+	}
+	newSecret := func() runtime.Object {
+		return &v1.Secret{}
+	}
+	gr := corev1.Resource("secret")
+	return &secretManager{
+		manager: manager.NewWatchBasedManager(listSecret, watchSecret, newSecret, gr, getSecretNames),
+	}
+}