Merge pull request #40795 from wojtek-t/use_caching_manager

Automatic merge from submit-queue (batch tested with PRs 40795, 40863)

Use caching secret manager in kubelet

I just found that this is in my local branch I'm using for testing, but not in master :)

pkg/kubelet/kubelet.go

@@ -415,7 +415,7 @@ func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *Kub
 	}
 	containerRefManager := kubecontainer.NewRefManager()
 
-	secretManager, err := secret.NewSimpleSecretManager(kubeClient)
+	secretManager, err := secret.NewCachingSecretManager(kubeClient)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize secret manager: %v", err)
 	}

pkg/kubelet/secret/BUILD

@@ -17,7 +17,10 @@ go_test(
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/clientset/fake:go_default_library",
         "//vendor:github.com/stretchr/testify/assert",
+        "//vendor:k8s.io/apimachinery/pkg/api/errors",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
+        "//vendor:k8s.io/apimachinery/pkg/runtime",
+        "//vendor:k8s.io/client-go/testing",
         "//vendor:k8s.io/client-go/util/clock",
     ],
 )

pkg/kubelet/secret/secret_manager.go

@@ -106,6 +106,9 @@ func newSecretStore(kubeClient clientset.Interface, clock clock.Clock, ttl time.
 }
 
 func isSecretOlder(newSecret, oldSecret *v1.Secret) bool {
+	if newSecret == nil || oldSecret == nil {
+		return false
+	}
 	newVersion, _ := storageetcd.Versioner.ObjectResourceVersion(newSecret)
 	oldVersion, _ := storageetcd.Versioner.ObjectResourceVersion(oldSecret)
 	return newVersion < oldVersion
@@ -178,18 +181,19 @@ func (s *secretStore) Get(namespace, name string) (*v1.Secret, error) {
 			util.FromApiserverCache(&opts)
 		}
 		secret, err := s.kubeClient.Core().Secrets(namespace).Get(name, opts)
-		// Update state, unless we got error different than "not-found".
-		if err == nil || apierrors.IsNotFound(err) {
-			// Ignore the update to the older version of a secret.
-			if data.secret == nil || secret == nil || !isSecretOlder(secret, data.secret) {
-				data.secret = secret
-				data.err = err
-				data.lastUpdateTime = s.clock.Now()
-			}
-		} else if data.secret == nil && data.err == nil {
-			// We have unitialized secretData - return current result.
+		if err != nil && !apierrors.IsNotFound(err) && data.secret == nil && data.err == nil {
+			// Couldn't fetch the latest secret, but there is no cached data to return.
+			// Return the fetch result instead.
 			return secret, err
 		}
+		if (err == nil && !isSecretOlder(secret, data.secret)) || apierrors.IsNotFound(err) {
+			// If the fetch succeeded with a newer version of the secret, or if the
+			// secret could not be found in the apiserver, update the cached data to
+			// reflect the current status.
+			data.secret = secret
+			data.err = err
+			data.lastUpdateTime = s.clock.Now()
+		}
 	}
 	return data.secret, data.err
 }

pkg/kubelet/secret/secret_manager_test.go

@@ -18,6 +18,7 @@ package secret
 
 import (
 	"fmt"
+	"reflect"
 	"strings"
 	"sync"
 	"testing"
@@ -26,7 +27,10 @@ import (
 	"k8s.io/kubernetes/pkg/api/v1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/util/clock"
 
 	"github.com/stretchr/testify/assert"
@@ -76,6 +80,35 @@ func TestSecretStore(t *testing.T) {
 	checkSecret(t, store, "ns4", "name4", false)
 }
 
+func TestSecretStoreDeletingSecret(t *testing.T) {
+	fakeClient := &fake.Clientset{}
+	store := newSecretStore(fakeClient, clock.RealClock{}, 0)
+	store.Add("ns", "name")
+
+	result := &v1.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name", ResourceVersion: "10"}}
+	fakeClient.AddReactor("get", "secrets", func(action core.Action) (bool, runtime.Object, error) {
+		return true, result, nil
+	})
+	secret, err := store.Get("ns", "name")
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(secret, result) {
+		t.Errorf("Unexpected secret: %v", secret)
+	}
+
+	fakeClient.PrependReactor("get", "secrets", func(action core.Action) (bool, runtime.Object, error) {
+		return true, &v1.Secret{}, apierrors.NewNotFound(v1.Resource("secret"), "name")
+	})
+	secret, err = store.Get("ns", "name")
+	if err == nil || !apierrors.IsNotFound(err) {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(secret, &v1.Secret{}) {
+		t.Errorf("Unexpected secret: %v", secret)
+	}
+}
+
 func TestSecretStoreGetAlwaysRefresh(t *testing.T) {
 	fakeClient := &fake.Clientset{}
 	fakeClock := clock.NewFakeClock(time.Now())