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,17 @@ 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() {
+			for {
+				go a.addClusterCacheInfoUpdateToQueue("periodic_sync")
+				time.Sleep(3 * time.Minute)
+			}
+		}()
+	}
+
 	if a.remote != nil {
 		a.remote.SetClientMode(a.mode)
 		// TODO: Right now, maintainConnection always returns nil. Revisit

agent/inbound.go

@@ -374,6 +374,12 @@ func (a *Agent) createApplication(incoming *v1alpha1.Application) (*v1alpha1.App
 		return created, nil
 	}
 
+	if a.mode == types.AgentModeManaged {
+		// When sync status changed to Synced, trigger a cluster cache info update event
+		// so that the principal can be informed of the cache update.
+		go a.addClusterCacheInfoUpdateToQueue("application_created")
+	}
+
 	return created, err
 }
 
@@ -448,8 +454,11 @@ 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)
+		// When sync status changed to Synced, trigger a cluster cache info update event
+		// so that the principal can be informed of the cache update.
+		go a.addClusterCacheInfoUpdateToQueue("application_deleted")
 	}
 
 	err = a.appManager.Unmanage(app.QualifiedName())

agent/outbound.go

@@ -15,11 +15,16 @@
 package agent
 
 import (
+	"errors"
+	"time"
+
+	"github.com/argoproj-labs/argocd-agent/internal/argocd/cluster"
 	"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"
 )
@@ -105,6 +110,13 @@ func (a *Agent) addAppUpdateToQueue(old *v1alpha1.Application, new *v1alpha1.App
 		WithField(logfields.SendQueueLen, q.Len()).
 		WithField(logfields.SendQueueName, defaultQueueName).
 		Debugf("Added event of type %s to send queue", eventType)
+
+	// When sync status changed to Synced, trigger a cluster cache info update event
+	// so that the principal can be informed of the cache update.
+	if a.mode == types.AgentModeManaged && old.Status.Sync.Status != new.Status.Sync.Status &&
+		new.Status.Sync.Status == v1alpha1.SyncStatusCodeSynced {
+		go a.addClusterCacheInfoUpdateToQueue("application_synced")
+	}
 }
 
 // addAppDeletionToQueue processes an application delete event originating from
@@ -261,3 +273,50 @@ 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(reason string) {
+	logCtx := log().WithFields(logrus.Fields{
+		"event":  "addClusterCacheInfoUpdateToQueue",
+		"reason": reason,
+	})
+
+	clusterServer := "https://kubernetes.default.svc"
+	var clusterInfo *v1alpha1.ClusterInfo
+	var err error
+
+	// If the reason is application_synced, application_created or application_deleted,
+	// we wait for 20 seconds to ensure that updated info is sent to principal.
+	// It is because the cluster cache is not updated immediately by Argo CD.
+	// Instead it is updated in next clusterInfoUpdater cycle.
+	if reason != "periodic_sync" {
+		time.Sleep(20 * time.Second)
+	}
+
+	// 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)
+	if err != nil {
+		if !errors.Is(err, cacheutil.ErrCacheMiss) {
+			logCtx.WithError(err).Errorf("Failed to get cluster info from cache")
+		}
+		return
+	}
+
+	// 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, clusterInfo)
+		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,
+			"reason":            reason,
+		}).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,13 @@ 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/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -371,3 +375,51 @@ 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("password123")},
+	}
+	_, 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(v1alpha1.ClusterInfo{
+		ApplicationsCount: 5,
+		CacheInfo: v1alpha1.ClusterCacheInfo{
+			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("periodic_sync")
+
+	// Should have an event in queue
+	require.Equal(t, 1, a.queues.SendQ(defaultQueueName).Len())
+}