feat: Populate cluster cache info in principal

Assisted by: Cursor

Signed-off-by: Jayendra Parsai <jparsai@redhat.com>

Author: Jayendra Parsai

agent/agent.go

@@ -353,6 +353,23 @@ func (a *Agent) Start(ctx context.Context) error {
 		go http.ListenAndServe(healthzAddr, nil)
 	}
 
+	// Start the background process of periodic sync of cluster cache info.
+	// This will send periodic updates of Application, Resource and API counts to principal.
+	if a.mode == types.AgentModeManaged {
+		go func() {
+			ticker := time.NewTicker(10 * time.Second)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ticker.C:
+					a.addClusterCacheInfoUpdateToQueue()
+				case <-a.context.Done():
+					return
+				}
+			}
+		}()
+	}
+
 	if a.remote != nil {
 		a.remote.SetClientMode(a.mode)
 		// TODO: Right now, maintainConnection always returns nil. Revisit

agent/inbound.go

@@ -448,7 +448,7 @@ func (a *Agent) deleteApplication(app *v1alpha1.Application) error {
 		return err
 	}
 
-	if a.mode == types.AgentModeManaged && err == nil {
+	if a.mode == types.AgentModeManaged {
 		appCache.DeleteApplicationSpec(app.UID, logCtx)
 	}
 

agent/outbound.go

@@ -15,15 +15,22 @@
 package agent
 
 import (
+	"errors"
+
+	"github.com/argoproj-labs/argocd-agent/internal/argocd/cluster"
+	"github.com/argoproj-labs/argocd-agent/internal/cache"
 	"github.com/argoproj-labs/argocd-agent/internal/event"
 	"github.com/argoproj-labs/argocd-agent/internal/logging/logfields"
 	"github.com/argoproj-labs/argocd-agent/internal/resources"
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
+	cacheutil "github.com/argoproj/argo-cd/v3/util/cache"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 )
 
+var eventSentAfterZeroApps = false
+
 // addAppCreationToQueue processes a new application event originating from the
 // AppInformer and puts it in the send queue.
 func (a *Agent) addAppCreationToQueue(app *v1alpha1.Application) {
@@ -261,3 +268,66 @@ func (a *Agent) addAppProjectDeletionToQueue(appProject *v1alpha1.AppProject) {
 	q.Add(a.emitter.AppProjectEvent(event.Delete, appProject))
 	logCtx.WithField(logfields.SendQueueLen, q.Len()).Debugf("Added appProject delete event to send queue")
 }
+
+// addClusterCacheInfoUpdateToQueue processes a cluster cache info update event
+// and puts it in the send queue.
+func (a *Agent) addClusterCacheInfoUpdateToQueue() {
+	logCtx := log().WithFields(logrus.Fields{
+		"event": "addClusterCacheInfoUpdateToQueue",
+	})
+
+	clusterServer := "https://kubernetes.default.svc"
+	var clusterInfo *v1alpha1.ClusterInfo
+	var err error
+
+	// Get the updated cluster info from agent's cache.
+	clusterInfo, err = cluster.GetClusterInfo(a.context, a.kubeClient.Clientset, a.namespace, clusterServer, a.redisProxyMsgHandler.redisAddress, cacheutil.RedisCompressionGZip, cache.SourceAgent)
+	if err != nil {
+		if !errors.Is(err, cacheutil.ErrCacheMiss) {
+			logCtx.WithError(err).Errorf("Failed to get cluster info from cache")
+		}
+		return
+	}
+
+	// If the cluster does not have any applications, Argo CD does not update the cluster cache info.
+	// Hence there is no need to send event to principal in this case, as information is same as last time.
+	// However, after application count changed to 0, we need to send one last event to principal to share the information after application deletion.
+	// For this, we use eventSentAfterZeroApps flag to track if the an event has been sent after application count changed to 0.
+
+	// If the cluster has applications, we reset the eventSentAfterZeroApps flag to false.
+	if clusterInfo.ApplicationsCount != 0 {
+		eventSentAfterZeroApps = false
+	}
+
+	// If the one event has been sent after application count changed to 0, then do nothing.
+	if eventSentAfterZeroApps {
+		return
+	}
+
+	// If the cluster has no applications, we set the eventSentAfterZeroApps flag to true.
+	// and send one event to principal to update the cluster cache info.
+	if clusterInfo.ApplicationsCount == 0 {
+		eventSentAfterZeroApps = true
+	}
+
+	// Send the event to principal to update the cluster cache info.
+	q := a.queues.SendQ(defaultQueueName)
+	if q != nil {
+		clusterInfoEvent := a.emitter.ClusterCacheInfoUpdateEvent(event.ClusterCacheInfoUpdate, &event.ClusterCacheInfo{
+			ApplicationsCount: clusterInfo.ApplicationsCount,
+			APIsCount:         clusterInfo.CacheInfo.APIsCount,
+			ResourcesCount:    clusterInfo.CacheInfo.ResourcesCount,
+		})
+
+		q.Add(clusterInfoEvent)
+		logCtx.WithFields(logrus.Fields{
+			"sendq_len":         q.Len(),
+			"sendq_name":        defaultQueueName,
+			"applicationsCount": clusterInfo.ApplicationsCount,
+			"apisCount":         clusterInfo.CacheInfo.APIsCount,
+			"resourcesCount":    clusterInfo.CacheInfo.ResourcesCount,
+		}).Infof("Added ClusterCacheInfoUpdate event to send queue")
+	} else {
+		logCtx.Error("Default queue not found, unable to send ClusterCacheInfoUpdate event")
+	}
+}

agent/outbound_test.go

@@ -3,9 +3,14 @@ package agent
 import (
 	"testing"
 
+	"github.com/alicebob/miniredis/v2"
+	"github.com/argoproj-labs/argocd-agent/internal/argocd/cluster"
 	"github.com/argoproj-labs/argocd-agent/internal/event"
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
+	"github.com/argoproj/argo-cd/v3/common"
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
+	cacheutil "github.com/argoproj/argo-cd/v3/util/cache"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -371,3 +376,49 @@ func Test_addAppProjectDeletionToQueue(t *testing.T) {
 		require.False(t, a.projectManager.IsManaged("test-project"))
 	})
 }
+
+func Test_addClusterCacheInfoUpdateToQueue(t *testing.T) {
+	a := newAgent(t)
+	a.remote.SetClientID("agent")
+	a.emitter = event.NewEventSource("principal")
+
+	miniRedis, err := miniredis.Run()
+	require.NoError(t, err)
+	require.NotNil(t, miniRedis)
+	defer miniRedis.Close()
+	err = WithRedisHost(miniRedis.Addr())(a)
+	require.NoError(t, err)
+
+	// Create the required Redis secret
+	secret := &corev1.Secret{
+		ObjectMeta: v1.ObjectMeta{Name: common.RedisInitialCredentials},
+		Data:       map[string][]byte{common.RedisInitialCredentialsKey: []byte(uuid.NewString())},
+	}
+	_, err = a.kubeClient.Clientset.CoreV1().Secrets("argocd").Create(a.context, secret, v1.CreateOptions{})
+	require.NoError(t, err)
+
+	// First populate the cache with dummy data
+	clusterMgr, err := cluster.NewManager(a.context, a.namespace, miniRedis.Addr(), cacheutil.RedisCompressionGZip, a.kubeClient.Clientset)
+	require.NoError(t, err)
+	err = clusterMgr.MapCluster("test-agent", &v1alpha1.Cluster{
+		Name:   "test-cluster",
+		Server: "https://kubernetes.default.svc",
+	})
+	require.NoError(t, err)
+
+	// Set dummy cluster cache stats
+	clusterMgr.SetClusterCacheStats(&event.ClusterCacheInfo{
+		ApplicationsCount: 5,
+		APIsCount:         10,
+		ResourcesCount:    100,
+	}, "test-agent")
+
+	// Should not have an event in queue
+	require.Equal(t, 0, a.queues.SendQ(defaultQueueName).Len())
+
+	// Add a cluster cache info update event to the queue
+	a.addClusterCacheInfoUpdateToQueue()
+
+	// Should have an event in queue
+	require.Equal(t, 1, a.queues.SendQ(defaultQueueName).Len())
+}

internal/argocd/cluster/cluster.go

@@ -15,20 +15,24 @@
 package cluster
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"time"
 
-	"github.com/redis/go-redis/v9"
+	"github.com/argoproj-labs/argocd-agent/internal/cache"
+	"github.com/argoproj-labs/argocd-agent/internal/event"
+	"github.com/sirupsen/logrus"
 
-	"github.com/argoproj/argo-cd/v3/common"
 	appv1 "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
 	cacheutil "github.com/argoproj/argo-cd/v3/util/cache"
-	appstatecache "github.com/argoproj/argo-cd/v3/util/cache/appstate"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
-// UpdateClusterConnectionInfo updates cluster info with connection state and time in mapped cluster
-func (m *Manager) UpdateClusterConnectionInfo(agentName, status string, t time.Time) {
+// SetAgentConnectionStatus updates cluster info with connection state and time in mapped cluster at principal.
+// This is called when the agent is connected or disconnected with the principal.
+func (m *Manager) SetAgentConnectionStatus(agentName, status appv1.ConnectionStatus, modifiedAt time.Time) {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
@@ -39,67 +43,135 @@ func (m *Manager) UpdateClusterConnectionInfo(agentName, status string, t time.T
 		return
 	}
 
-	// Create Redis client and cache
-	redisOptions := &redis.Options{Addr: m.redisAddress}
-
-	if err := common.SetOptionalRedisPasswordFromKubeConfig(m.ctx, m.kubeclient, m.namespace, redisOptions); err != nil {
-		log().Errorf("Failed to fetch & set redis password for namespace %s: %v", m.namespace, err)
-	}
-	redisClient := redis.NewClient(redisOptions)
-
-	cache := appstatecache.NewCache(cacheutil.NewCache(
-		cacheutil.NewRedisCache(redisClient, time.Minute, m.redisCompressionType)), time.Minute)
-
 	state := "disconnected"
 	if status == appv1.ConnectionStatusSuccessful {
 		state = "connected"
 	}
 
-	// update the info
-	if err := cache.SetClusterInfo(cluster.Server,
+	// Update the cluster connection state and time in mapped cluster at principal.
+	if err := m.setClusterInfo(cluster.Server, agentName, cluster.Name,
 		&appv1.ClusterInfo{
 			ConnectionState: appv1.ConnectionState{
 				Status:     status,
 				Message:    fmt.Sprintf("Agent: '%s' is %s with principal", agentName, state),
-				ModifiedAt: &metav1.Time{Time: t}}},
-	); err != nil {
-		log().Errorf("Failed to update connection info in Cluster: '%s' mapped with Agent: '%s'. Error: %v", cluster.Name, agentName, err)
+				ModifiedAt: &metav1.Time{Time: modifiedAt},
+			},
+		}); err != nil {
+		log().Errorf("failed to refresh connection info in cluster: '%s' mapped with agent: '%s'. Error: %v", cluster.Name, agentName, err)
 		return
 	}
 
 	log().Infof("Updated connection status to '%s' in Cluster: '%s' mapped with Agent: '%s'", status, cluster.Name, agentName)
 }
 
-// refreshClusterConnectionInfo gets latest cluster info from cache and re-saves it to avoid deletion of info
-// by ArgoCD after cache expiration time duration (i.e. 10 Minute)
-func (m *Manager) refreshClusterConnectionInfo() {
+// refreshClusterInfo gets latest cluster info from cache and re-saves it to avoid deletion of info
+// by Argo CD after cache expiration time duration (i.e. 10 minutes)
+func (m *Manager) refreshClusterInfo() {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
-	// iterate through all clusters
+	// Iterate through all clusters.
 	for agentName, cluster := range m.clusters {
-		// Create Redis client and cache
-		redisOptions := &redis.Options{Addr: m.redisAddress}
+		clusterInfo, err := GetClusterInfo(m.ctx, m.kubeclient, m.namespace, cluster.Server, m.redisAddress, m.redisCompressionType, cache.SourcePrincipal)
+		if err != nil {
+			if !errors.Is(err, cacheutil.ErrCacheMiss) {
+				log().Errorf("failed to get connection info from cluster: '%s' mapped with agent: '%s'. Error: %v", cluster.Name, agentName, err)
+			}
+			continue
+		}
 
-		if err := common.SetOptionalRedisPasswordFromKubeConfig(m.ctx, m.kubeclient, m.namespace, redisOptions); err != nil {
-			log().Errorf("Failed to fetch & set redis password for namespace %s: %v", m.namespace, err)
+		// Re-save same info.
+		if err := m.setClusterInfo(cluster.Server, agentName, cluster.Name, clusterInfo); err != nil {
+			log().Errorf("failed to refresh connection info in cluster: '%s' mapped with agent: '%s'. Error: %v", cluster.Name, agentName, err)
+			continue
 		}
-		redisClient := redis.NewClient(redisOptions)
+	}
+}
 
-		cache := appstatecache.NewCache(cacheutil.NewCache(
-			cacheutil.NewRedisCache(redisClient, time.Minute, m.redisCompressionType)), time.Minute)
+// SetClusterCacheStats updates cluster cache info with Application, Resource and API counts in principal.
+// This is called when principal receives clusterCacheInfoUpdate event from agent.
+func (m *Manager) SetClusterCacheStats(clusterInfo *event.ClusterCacheInfo, agentName string) error {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
 
-		// fetch latest info
-		clusterInfo := &appv1.ClusterInfo{}
-		if err := cache.GetClusterInfo(cluster.Server, clusterInfo); err != nil {
-			log().Errorf("Failed to get connection info from Cluster: '%s' mapped with Agent: '%s'. Error: %v", cluster.Name, agentName, err)
-			return
-		}
+	// Check if we have a mapping for the requested agent.
+	cluster := m.mapping(agentName)
+	if cluster == nil {
+		log().Errorf("agent %s is not mapped to any cluster", agentName)
+		return fmt.Errorf("agent %s is not mapped to any cluster", agentName)
+	}
+
+	// Get existing cluster info to preserve cluster connection status
+	existingClusterInfo, err := GetClusterInfo(m.ctx, m.kubeclient, m.namespace, cluster.Server, m.redisAddress, m.redisCompressionType, cache.SourcePrincipal)
+	if err != nil && !errors.Is(err, cacheutil.ErrCacheMiss) {
+		log().Errorf("failed to get existing cluster info for cluster: '%s' mapped with agent: '%s'. Error: %v", cluster.Name, agentName, err)
+		// Continue with default values if we can't get existing info
+	}
+
+	// Create new cluster info with cache stats
+	newClusterInfo := &appv1.ClusterInfo{
+		ApplicationsCount: clusterInfo.ApplicationsCount,
+		CacheInfo: appv1.ClusterCacheInfo{
+			APIsCount:      clusterInfo.APIsCount,
+			ResourcesCount: clusterInfo.ResourcesCount,
+		},
+	}
 
-		// re-save same info
-		if err := cache.SetClusterInfo(cluster.Server, clusterInfo); err != nil {
-			log().Errorf("Failed to refresh connection info in Cluster: '%s' mapped with Agent: '%s'. Error: %v", cluster.Name, agentName, err)
-			return
+	// Preserve existing cluster connection status
+	if existingClusterInfo != nil {
+		newClusterInfo.ConnectionState = existingClusterInfo.ConnectionState
+		if existingClusterInfo.CacheInfo.LastCacheSyncTime != nil {
+			newClusterInfo.CacheInfo.LastCacheSyncTime = existingClusterInfo.CacheInfo.LastCacheSyncTime
 		}
 	}
+
+	// Set the info in mapped cluster at principal.
+	if err := m.setClusterInfo(cluster.Server, agentName, cluster.Name, newClusterInfo); err != nil {
+		log().Errorf("failed to update cluster cache stats in cluster: '%s' mapped with agent: '%s'. Error: %v", cluster.Name, agentName, err)
+		return err
+	}
+
+	log().WithFields(logrus.Fields{
+		"applicationsCount": clusterInfo.ApplicationsCount,
+		"apisCount":         clusterInfo.APIsCount,
+		"resourcesCount":    clusterInfo.ResourcesCount,
+		"cluster":           cluster.Name,
+		"agent":             agentName,
+	}).Infof("Updated cluster cache stats in cluster.")
+
+	return nil
+}
+
+// setClusterInfo saves the given ClusterInfo in the cache.
+func (m *Manager) setClusterInfo(clusterServer, agentName, clusterName string, clusterInfo *appv1.ClusterInfo) error {
+
+	// Get cluster cache instance from redis.
+	clusterCache, err := cache.GetCacheInstance(m.ctx, m.kubeclient, m.namespace, m.redisAddress, m.redisCompressionType, cache.SourcePrincipal)
+	if err != nil {
+		return fmt.Errorf("failed to get cluster cache instance: %v", err)
+	}
+
+	// Save the given cluster info in cache.
+	if err := clusterCache.SetClusterInfo(clusterServer, clusterInfo); err != nil {
+		return fmt.Errorf("failed to refresh connection info in cluster: '%s' mapped with agent: '%s': %v", clusterName, agentName, err)
+	}
+	return nil
+}
+
+// GetClusterInfo retrieves the ClusterInfo for a given cluster from the cache.
+func GetClusterInfo(ctx context.Context, kubeclient kubernetes.Interface, namespace,
+	clusterServer, redisAddress string, redisCompressionType cacheutil.RedisCompressionType, role string) (*appv1.ClusterInfo, error) {
+
+	// Get cluster cache instance from redis.
+	clusterCache, err := cache.GetCacheInstance(ctx, kubeclient, namespace, redisAddress, redisCompressionType, role)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get cluster cache instance: %v", err)
+	}
+
+	clusterInfo := &appv1.ClusterInfo{}
+	// Fetch the cluster info from cache.
+	if err := clusterCache.GetClusterInfo(clusterServer, clusterInfo); err != nil {
+		return nil, err
+	}
+	return clusterInfo, nil
 }