[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

@@ -10,4 +10,10 @@ import "github.com/elastic/elastic-agent-autodiscover/kubernetes"
 type Config struct {
 	KubeConfig        string                       `config:"kube_config"`
 	KubeClientOptions kubernetes.KubeClientOptions `config:"kube_client_options"`
+
+	TTL int `config:"ttl"`
+}
+
+func (c *Config) InitDefaults() {
+	c.TTL = 1
 }

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,9 @@ type contextProviderK8sSecrets struct {
 
 	clientMx sync.Mutex
 	client   k8sclient.Interface
+
+	secretsCacheMx sync.Mutex
+	secretsCache   map[string]string
 }
 
 // ContextProviderBuilder builds the context provider.
@@ -46,19 +50,102 @@ 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]string),
 	}, 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()
+	p.client = client
+	p.clientMx.Unlock()
+	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) {
+	d := time.Duration(p.config.TTL) * time.Minute
+	timer := time.NewTimer(d)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			p.updateCache()
+			timer.Reset(d)
+		}
+	}
+}
+
+func (p *contextProviderK8sSecrets) updateCache() {
+	p.secretsCacheMx.Lock()
+	for name := range p.secretsCache {
+		newValue, ok := p.fetchSecret(name)
+
+		// remove the secret from the cache
+		if !ok {
+			delete(p.secretsCache, name)
+		} else {
+			p.secretsCache[name] = newValue
+		}
+	}
+	p.secretsCacheMx.Unlock()
+}
+
+func (p *contextProviderK8sSecrets) getFromCache(key string) (string, bool) {
+	p.secretsCacheMx.Lock()
+	value, ok := p.secretsCache[key]
+	p.secretsCacheMx.Unlock()
+
+	// if value is still not present in cache, it is possible we haven't tried to fetch it yet
+	if !ok {
+		value, ok = p.addToCache(key)
+	}
+	return value, ok
+}
+
+func (p *contextProviderK8sSecrets) addToCache(key string) (string, bool) {
+	value, ok := p.fetchSecret(key)
+	if ok {
+		p.secretsCacheMx.Lock()
+		p.secretsCache[key] = value
+		p.secretsCacheMx.Unlock()
+	}
+	return value, ok
+}
+
+func (p *contextProviderK8sSecrets) fetchSecret(key string) (string, bool) {
 	p.clientMx.Lock()
 	client := p.client
 	p.clientMx.Unlock()
 	if client == nil {
 		return "", false
 	}
+
+	// key = "kubernetes_secrets.somenamespace.somesecret.value"
 	tokens := strings.Split(key, ".")
 	if len(tokens) > 0 && tokens[0] != "kubernetes_secrets" {
 		return "", false
@@ -87,26 +174,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
 }