[Kubernetes secret provider] Send signal to agent if cache is updated (#4371)

* Send signal if cache is updated

* Signal tests

* Add changelog fragment and fix typos

* Update changelog/fragments/1709824109-k8s-secret-provider-trigger-signal.yaml

Co-authored-by: Blake Rouse <blake.rouse@elastic.co>

* Fix racing condition

---------

Co-authored-by: Blake Rouse <blake.rouse@elastic.co>

Author: constanca-m

changelog/fragments/1709824109-k8s-secret-provider-trigger-signal.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: Kubernetes secrets provider has been improved to update a kubernetes secret  when the secret value changes.
+
+# 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/4371
+
+# 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/4168

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

@@ -86,7 +86,7 @@ func (p *contextProviderK8sSecrets) Run(ctx context.Context, comm corecomp.Conte
 	p.clientMx.Unlock()
 
 	if !p.config.DisableCache {
-		go p.updateSecrets(ctx)
+		go p.updateSecrets(ctx, comm)
 	}
 
 	<-comm.Done()
@@ -102,28 +102,39 @@ func getK8sClient(kubeconfig string, opt kubernetes.KubeClientOptions) (k8sclien
 }
 
 // Update the secrets in the cache every RefreshInterval
-func (p *contextProviderK8sSecrets) updateSecrets(ctx context.Context) {
+func (p *contextProviderK8sSecrets) updateSecrets(ctx context.Context, comm corecomp.ContextProviderComm) {
 	timer := time.NewTimer(p.config.RefreshInterval)
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-timer.C:
-			p.updateCache()
+			updatedCache := p.updateCache()
+			if updatedCache {
+				p.logger.Info("Secrets cache was updated, the agent will be notified.")
+				comm.Signal()
+			}
 			timer.Reset(p.config.RefreshInterval)
 		}
 	}
 }
 
 // mergeWithCurrent merges the updated map with the cache map.
 // This function needs to be called between the mutex lock for the map.
-func (p *contextProviderK8sSecrets) mergeWithCurrent(updatedMap map[string]*secretsData) map[string]*secretsData {
+func (p *contextProviderK8sSecrets) mergeWithCurrent(updatedMap map[string]*secretsData) (map[string]*secretsData, bool) {
 	merged := make(map[string]*secretsData)
+	updatedCache := false
 
 	for name, data := range p.secretsCache {
 		diff := time.Since(data.lastAccess)
 		if diff < p.config.TTLDelete {
 			merged[name] = data
+			// Check if this key is part of the updatedMap. If it is not, we know the secrets cache was updated,
+			// and we need to signal that.
+			_, ok := updatedMap[name]
+			if !ok {
+				updatedCache = true
+			}
 		}
 	}
 
@@ -132,14 +143,20 @@ func (p *contextProviderK8sSecrets) mergeWithCurrent(updatedMap map[string]*secr
 		// it could have been updated when trying to fetch the secret at the same time we are running update cache.
 		// In that case, we only update the value.
 		if _, ok := merged[name]; ok {
-			merged[name].value = data.value
+			if merged[name].value != data.value {
+				merged[name].value = data.value
+				updatedCache = true
+			}
 		}
 	}
 
-	return merged
+	return merged, updatedCache
 }
 
-func (p *contextProviderK8sSecrets) updateCache() {
+func (p *contextProviderK8sSecrets) updateCache() bool {
+	// Keep track whether the cache had values changing, so we can notify the agent
+	updatedCache := false
+
 	// 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)
@@ -152,6 +169,8 @@ func (p *contextProviderK8sSecrets) updateCache() {
 	}
 	p.secretsCacheMx.RUnlock()
 
+	// The only way to update an entry in the cache is through the last access time (to delete the key)
+	// or if the value gets updated.
 	for name, data := range copyMap {
 		diff := time.Since(data.lastAccess)
 		if diff < p.config.TTLDelete {
@@ -162,17 +181,24 @@ func (p *contextProviderK8sSecrets) updateCache() {
 					lastAccess: data.lastAccess,
 				}
 				cacheTmp[name] = newData
+				if value != data.value {
+					updatedCache = true
+				}
 			}
-
+		} else {
+			updatedCache = true
 		}
 	}
 
 	// While the cache was updated, it is possible that some secret was added through another go routine.
 	// We need to merge the updated map with the current cache map to catch the new entries and avoid
 	// loss of data.
+	var updated bool
 	p.secretsCacheMx.Lock()
-	p.secretsCache = p.mergeWithCurrent(cacheTmp)
+	p.secretsCache, updated = p.mergeWithCurrent(cacheTmp)
 	p.secretsCacheMx.Unlock()
+
+	return updatedCache || updated
 }
 
 func (p *contextProviderK8sSecrets) getFromCache(key string) (string, bool) {