util: cache the controller publish secret

This patch introduces a caching mechanism based on the
shared informers for ControllerPublish secrets.

Signed-off-by: Niraj Yadav <niryadav@redhat.com>

Author: black-dragon74

internal/controller/controller.go

@@ -18,6 +18,7 @@ package controller
 import (
 	"fmt"
 
+	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
 
 	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1"
@@ -111,5 +112,11 @@ func Start(config Config) error {
 		log.ErrorLogMsg("failed to start manager %s", err)
 	}
 
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	if err := k8s.StartSecretWatcher(stopCh); err != nil {
+		log.ErrorLogMsg("failed to start secret watcher: %v", err)
+	}
+
 	return err
 }

internal/util/k8s/secrets.go

@@ -18,15 +18,90 @@ package k8s
 
 import (
 	"context"
+	"crypto/sha1"
+	"encoding/hex"
 	"fmt"
+	"sync"
+	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
 )
 
+type cachedSecret struct {
+	data map[string]string
+}
+
+var (
+	secretCache = make(map[string]cachedSecret)
+	cacheLock   sync.RWMutex
+)
+
+// cacheKey is a helper to create sha1 hash based on a secret's name
+// and namespace. The hash is then used as an index for secretCache
+func cacheKey(ns, name string) string {
+	h := sha1.New()
+	h.Write([]byte(ns + "/" + name))
+
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// StartSecretWatcher handles events on Secrets informer and updates
+// or invalidates the secretCache accordingly.
+// It should preferably be started early in the bootstraping process.
+func StartSecretWatcher(stopCh chan struct{}) error {
+	client, err := NewK8sClient()
+	if err != nil {
+		return fmt.Errorf("failed to connect to Kubernetes: %w", err)
+	}
+
+	factory := informers.NewSharedInformerFactoryWithOptions(
+		client,
+		time.Minute*10, // FIXME: We could optionally restrict this to specific NS.
+	)
+
+	informer := factory.Core().V1().Secrets().Informer()
+	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		UpdateFunc: func(oldObj, newObj any) {
+			newSecret := newObj.(*corev1.Secret)
+			key := cacheKey(newSecret.Namespace, newSecret.Name)
+
+			cacheLock.Lock()
+			delete(secretCache, key)
+			cacheLock.Unlock()
+		},
+		DeleteFunc: func(obj any) {
+			secret := obj.(*corev1.Secret)
+			key := cacheKey(secret.Namespace, secret.Name)
+
+			cacheLock.Lock()
+			delete(secretCache, key)
+			cacheLock.Unlock()
+		},
+	})
+
+	go informer.Run(stopCh)
+	return nil
+}
+
 // GetSecret retrieves a Kubernetes Secret by its name and namespace.
 // It returns a map of key-value pairs contained in the Secret's data.
 // If the Secret does not exist or cannot be accessed, it returns an error.
 func GetSecret(secretName, secretNamespace string) (map[string]string, error) {
+	key := cacheKey(secretNamespace, secretName)
+
+	// Return from cache if present
+	cacheLock.RLock()
+	if entry, found := secretCache[key]; found {
+		cacheLock.RUnlock()
+
+		return entry.data, nil
+	}
+	cacheLock.RUnlock()
+
+	// The entry was not found in cache, fetch it from the API server
 	client, err := NewK8sClient()
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to Kubernetes: %w", err)
@@ -37,10 +112,17 @@ func GetSecret(secretName, secretNamespace string) (map[string]string, error) {
 		return nil, fmt.Errorf("failed to get secret %q in namespace %q information: %w", secretName, secretNamespace, err)
 	}
 
-	secretData := map[string]string{}
+	secretData := make(map[string]string, len(secret.Data))
 	for key, value := range secret.Data {
 		secretData[key] = string(value)
 	}
 
+	// Cache the fetched secret
+	cacheLock.Lock()
+	secretCache[key] = cachedSecret{
+		data: secretData,
+	}
+	cacheLock.Unlock()
+
 	return secretData, nil
 }