[Kubernetes secret provider] Add cache for the secrets

changelog/fragments/1701091034-add-cache-for-secrets.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: add cache for secrets when using kubernetes secret provider
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/3822
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/elastic-agent/issues/3594

internal/pkg/composable/providers/kubernetessecrets/config.go

@@ -4,10 +4,22 @@
 
 package kubernetessecrets
 
-import "github.com/elastic/elastic-agent-autodiscover/kubernetes"
+import (
+	"time"
+
+	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
+)
 
 // Config for kubernetes provider
 type Config struct {
 	KubeConfig        string                       `config:"kube_config"`
 	KubeClientOptions kubernetes.KubeClientOptions `config:"kube_client_options"`
+
+	TTLUpdate time.Duration `config:"ttl_update"`
+	TTLDelete time.Duration `config:"ttl_delete"`
+}
+
+func (c *Config) InitDefaults() {
+	c.TTLUpdate = 60 * time.Second
+	c.TTLDelete = 1 * time.Hour
 }

internal/pkg/composable/providers/kubernetessecrets/kubernetes_secrets.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"strings"
 	"sync"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sclient "k8s.io/client-go/kubernetes"
@@ -33,6 +34,14 @@ type contextProviderK8sSecrets struct {
 
 	clientMx sync.Mutex
 	client   k8sclient.Interface
+
+	secretsCacheMx sync.RWMutex
+	secretsCache   map[string]*secretsData
+}
+
+type secretsData struct {
+	value      string
+	lastAccess time.Time
 }
 
 // ContextProviderBuilder builds the context provider.
@@ -46,38 +55,152 @@ func ContextProviderBuilder(logger *logger.Logger, c *config.Config, managed boo
 		return nil, errors.New(err, "failed to unpack configuration")
 	}
 	return &contextProviderK8sSecrets{
-		logger: logger,
-		config: &cfg,
+		logger:       logger,
+		config:       &cfg,
+		secretsCache: make(map[string]*secretsData),
 	}, nil
 }
 
 func (p *contextProviderK8sSecrets) Fetch(key string) (string, bool) {
-	// key = "kubernetes_secrets.somenamespace.somesecret.value"
+	return p.getFromCache(key)
+}
+
+// Run initializes the k8s secrets context provider.
+func (p *contextProviderK8sSecrets) Run(ctx context.Context, comm corecomp.ContextProviderComm) error {
+	client, err := getK8sClientFunc(p.config.KubeConfig, p.config.KubeClientOptions)
+	if err != nil {
+		p.logger.Debugf("Kubernetes_secrets provider skipped, unable to connect: %s", err)
+		return nil
+	}
 	p.clientMx.Lock()
-	client := p.client
+	p.client = client
 	p.clientMx.Unlock()
-	if client == nil {
-		return "", false
+	go p.updateSecrets(ctx)
+
+	<-comm.Done()
+
+	p.clientMx.Lock()
+	p.client = nil
+	p.clientMx.Unlock()
+	return comm.Err()
+}
+
+func getK8sClient(kubeconfig string, opt kubernetes.KubeClientOptions) (k8sclient.Interface, error) {
+	return kubernetes.GetKubernetesClient(kubeconfig, opt)
+}
+
+// Update the secrets in the cache every TTL minutes
+func (p *contextProviderK8sSecrets) updateSecrets(ctx context.Context) {
+	timer := time.NewTimer(p.config.TTLUpdate)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			p.updateCache()
+			timer.Reset(p.config.TTLUpdate)
+		}
 	}
+}
+
+func (p *contextProviderK8sSecrets) updateCache() {
+	p.secretsCacheMx.Lock()
+	// deleting entries does not free the memory, so we need to create a new map
+	// to place the secrets we want to keep
+	cacheTmp := make(map[string]*secretsData)
+	for name, data := range p.secretsCache {
+		diff := time.Since(data.lastAccess)
+		if diff > p.config.TTLDelete {
+			delete(p.secretsCache, name)
+		} else {
+			newValue, ok := p.fetchSecret(name)
+			if !ok {
+				delete(p.secretsCache, name)
+			} else {
+				data.value = newValue
+				cacheTmp[name] = data
+			}
+		}
+	}
+	p.secretsCache = cacheTmp
+	p.secretsCacheMx.Unlock()
+}
+
+func (p *contextProviderK8sSecrets) getFromCache(key string) (string, bool) {
+	p.secretsCacheMx.RLock()
+	_, ok := p.secretsCache[key]
+	p.secretsCacheMx.RUnlock()
+
+	// if value is still not present in cache, it is possible we haven't tried to fetch it yet
+	if !ok {
+		data, ok := p.addToCache(key)
+		// if it was not possible to fetch the secret, return
+		if !ok {
+			return data.value, ok
+		}
+	}
+
+	var pass string
+	p.secretsCacheMx.Lock()
+	data, ok := p.secretsCache[key]
+	data.lastAccess = time.Now()
+	pass = data.value
+	p.secretsCacheMx.Unlock()
+	return pass, ok
+}
+
+func (p *contextProviderK8sSecrets) addToCache(key string) (secretsData, bool) {
+	// Make sure the key has the expected format "kubernetes_secrets.somenamespace.somesecret.value"
 	tokens := strings.Split(key, ".")
 	if len(tokens) > 0 && tokens[0] != "kubernetes_secrets" {
-		return "", false
+		return secretsData{
+			value: "",
+		}, false
 	}
 	if len(tokens) != 4 {
 		p.logger.Debugf(
 			"not valid secret key: %v. Secrets should be of the following format %v",
 			key,
 			"kubernetes_secrets.somenamespace.somesecret.value",
 		)
+		return secretsData{
+			value: "",
+		}, false
+	}
+
+	value, ok := p.fetchSecret(key)
+	data := secretsData{
+		value: value,
+	}
+	if ok {
+		p.secretsCacheMx.Lock()
+		p.secretsCache[key] = &data
+		p.secretsCacheMx.Unlock()
+	}
+	return data, ok
+}
+
+func (p *contextProviderK8sSecrets) fetchSecret(key string) (string, bool) {
+	p.clientMx.Lock()
+	client := p.client
+	p.clientMx.Unlock()
+	if client == nil {
 		return "", false
 	}
+
+	tokens := strings.Split(key, ".")
+	// key has the format "kubernetes_secrets.somenamespace.somesecret.value"
+	// This function is only called from:
+	// - addToCache, where we already validated that the key has the right format.
+	// - updateCache, where the results are only added to the cache through addToCache
+	// Because of this we no longer need to validate the key
 	ns := tokens[1]
 	secretName := tokens[2]
 	secretVar := tokens[3]
 
-	secretIntefrace := client.CoreV1().Secrets(ns)
+	secretInterface := client.CoreV1().Secrets(ns)
 	ctx := context.TODO()
-	secret, err := secretIntefrace.Get(ctx, secretName, metav1.GetOptions{})
+	secret, err := secretInterface.Get(ctx, secretName, metav1.GetOptions{})
 	if err != nil {
 		p.logger.Errorf("Could not retrieve secret from k8s API: %v", err)
 		return "", false
@@ -87,26 +210,6 @@ func (p *contextProviderK8sSecrets) Fetch(key string) (string, bool) {
 		return "", false
 	}
 	secretString := secret.Data[secretVar]
-	return string(secretString), true
-}
 
-// Run initializes the k8s secrets context provider.
-func (p *contextProviderK8sSecrets) Run(ctx context.Context, comm corecomp.ContextProviderComm) error {
-	client, err := getK8sClientFunc(p.config.KubeConfig, p.config.KubeClientOptions)
-	if err != nil {
-		p.logger.Debugf("Kubernetes_secrets provider skipped, unable to connect: %s", err)
-		return nil
-	}
-	p.clientMx.Lock()
-	p.client = client
-	p.clientMx.Unlock()
-	<-comm.Done()
-	p.clientMx.Lock()
-	p.client = nil
-	p.clientMx.Unlock()
-	return comm.Err()
-}
-
-func getK8sClient(kubeconfig string, opt kubernetes.KubeClientOptions) (k8sclient.Interface, error) {
-	return kubernetes.GetKubernetesClient(kubeconfig, opt)
+	return string(secretString), true
 }

internal/pkg/composable/providers/kubernetessecrets/kubernetes_secrets_test.go

@@ -138,3 +138,119 @@ func Test_K8sSecretsProvider_FetchWrongSecret(t *testing.T) {
 	assert.False(t, found)
 	assert.EqualValues(t, val, "")
 }
+
+func Test_K8sSecretsProvider_Check_TTL(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
+
+	ttlDelete, err := time.ParseDuration("1s")
+	require.NoError(t, err)
+
+	ttlUpdate, err := time.ParseDuration("100ms")
+	require.NoError(t, err)
+
+	secret := &v1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Secret",
+			APIVersion: "apps/v1beta1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testing_secret",
+			Namespace: ns,
+		},
+		Data: map[string][]byte{
+			"secret_value": []byte(pass),
+		},
+	}
+	_, err = client.CoreV1().Secrets(ns).Create(context.Background(), secret, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	logger := logp.NewLogger("test_k8s_secrets")
+
+	c := map[string]interface{}{
+		"ttl_update": ttlUpdate,
+		"ttl_delete": ttlDelete,
+	}
+	cfg, err := config.NewConfigFrom(c)
+	require.NoError(t, err)
+
+	p, err := ContextProviderBuilder(logger, cfg, true)
+	require.NoError(t, err)
+
+	fp, _ := p.(*contextProviderK8sSecrets)
+
+	getK8sClientFunc = func(kubeconfig string, opt kubernetes.KubeClientOptions) (k8sclient.Interface, error) {
+		return client, nil
+	}
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	comm := ctesting.NewContextComm(ctx)
+
+	go func() {
+		_ = fp.Run(ctx, comm)
+	}()
+
+	for {
+		fp.clientMx.Lock()
+		client := fp.client
+		fp.clientMx.Unlock()
+		if client != nil {
+			break
+		}
+		<-time.After(10 * time.Millisecond)
+	}
+
+	// Secret cache should be empty at start
+	fp.secretsCacheMx.Lock()
+	assert.Equal(t, 0, len(fp.secretsCache))
+	fp.secretsCacheMx.Unlock()
+
+	key := "kubernetes_secrets.test_namespace.testing_secret.secret_value"
+
+	// Secret should be in the cache after this call
+	val, found := fp.Fetch(key)
+	assert.True(t, found)
+	assert.Equal(t, val, pass)
+	fp.secretsCacheMx.RLock()
+	assert.Equal(t, len(fp.secretsCache), 1)
+	assert.NotNil(t, fp.secretsCache[key])
+	assert.NotZero(t, fp.secretsCache[key].lastAccess)
+	fp.secretsCacheMx.RUnlock()
+
+	// Update the secret and check after TTL time, the secret value is correct
+	newPass := "new-pass"
+	secret = &v1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Secret",
+			APIVersion: "apps/v1beta1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testing_secret",
+			Namespace: ns,
+		},
+		Data: map[string][]byte{
+			"secret_value": []byte(newPass),
+		},
+	}
+	_, err = client.CoreV1().Secrets(ns).Update(context.Background(), secret, metav1.UpdateOptions{})
+	require.NoError(t, err)
+
+	// wait for ttl update
+	<-time.After(ttlUpdate)
+	assert.Eventuallyf(t, func() bool {
+		val, found = fp.Fetch(key)
+		return found && val == newPass
+	}, ttlUpdate*3, ttlUpdate, "Failed to update the secret value after TTL update has passed.")
+
+	// After TTL delete, secret should no longer be found in cache since it was never
+	// fetched during that time
+	<-time.After(ttlDelete)
+	assert.Eventuallyf(t, func() bool {
+		fp.secretsCacheMx.RLock()
+		size := len(fp.secretsCache)
+		fp.secretsCacheMx.RUnlock()
+		return size == 0
+	}, ttlDelete*3, ttlDelete, "Failed to delete the secret after TTL delete has passed.")
+
+}