feat: Populate cluster cache info in principal (#513)

Signed-off-by: Jayendra Parsai <jparsai@redhat.com>
Co-authored-by: Jayendra Parsai <jparsai@jparsai-thinkpadp1gen4i.remote.csb>

Author: jparsai

agent/agent.go

@@ -22,6 +22,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/argoproj-labs/argocd-agent/internal/argocd/cluster"
 	"github.com/argoproj-labs/argocd-agent/internal/backend"
 	kubeapp "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/application"
 	kubeappproject "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/appproject"
@@ -47,6 +48,8 @@ import (
 
 	appCache "github.com/argoproj-labs/argocd-agent/internal/cache"
 	"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"
 	ty "k8s.io/apimachinery/pkg/types"
 )
 
@@ -94,6 +97,9 @@ type Agent struct {
 
 	// enableResourceProxy determines if the agent should proxy resources to the principal
 	enableResourceProxy bool
+
+	cacheRefreshInterval time.Duration
+	clusterCache         *appstatecache.Cache
 }
 
 const defaultQueueName = "default"
@@ -273,6 +279,13 @@ func NewAgent(ctx context.Context, client *kube.KubernetesClient, namespace stri
 		connMap: map[string]connectionEntry{},
 	}
 
+	clusterCache, err := cluster.NewClusterCacheInstance(ctx, client.Clientset,
+		a.namespace, a.redisProxyMsgHandler.redisAddress, cacheutil.RedisCompressionGZip)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create cluster cache instance: %v", err)
+	}
+	a.clusterCache = clusterCache
+
 	return a, nil
 }
 
@@ -353,6 +366,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(a.cacheRefreshInterval)
+			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/agent_test.go

@@ -29,7 +29,7 @@ import (
 
 func newAgent(t *testing.T) *Agent {
 	t.Helper()
-	kubec := kube.NewKubernetesFakeClientWithApps()
+	kubec := kube.NewKubernetesFakeClientWithApps("argocd")
 	remote, err := client.NewRemote("127.0.0.1", 8080)
 	require.NoError(t, err)
 	agent, err := NewAgent(context.TODO(), kubec, "argocd", WithRemote(remote))
@@ -38,7 +38,7 @@ func newAgent(t *testing.T) *Agent {
 }
 
 func Test_NewAgent(t *testing.T) {
-	kubec := kube.NewKubernetesFakeClientWithApps()
+	kubec := kube.NewKubernetesFakeClientWithApps("agent")
 	agent, err := NewAgent(context.TODO(), kubec, "agent", WithRemote(&client.Remote{}))
 	require.NotNil(t, agent)
 	require.NoError(t, err)

agent/inbound.go

@@ -457,7 +457,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/inbound_test.go

@@ -103,7 +103,7 @@ func Test_ProcessIncomingAppWithUIDMismatch(t *testing.T) {
 				Name:       "owner",
 				UID:        "uid-1",
 			},
-		},	
+		},
 	}}
 
 	// oldApp is the app that is already present on the agent

agent/options.go

@@ -16,6 +16,7 @@ package agent
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/argoproj-labs/argocd-agent/pkg/client"
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
@@ -99,3 +100,10 @@ func WithEnableResourceProxy(enable bool) AgentOption {
 		return nil
 	}
 }
+
+func WithCacheRefreshInterval(interval time.Duration) AgentOption {
+	return func(o *Agent) error {
+		o.cacheRefreshInterval = interval
+		return nil
+	}
+}

agent/outbound.go

@@ -15,11 +15,14 @@
 package agent
 
 import (
+	"errors"
+
 	"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"
 )
@@ -261,3 +264,43 @@ 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"
+	clusterInfo := &v1alpha1.ClusterInfo{}
+
+	// Get the updated cluster info from agent's cache.
+	if err := a.clusterCache.GetClusterInfo(clusterServer, clusterInfo); 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, &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

@@ -1,11 +1,17 @@
 package agent
 
 import (
+	"context"
 	"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/client"
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
+	"github.com/argoproj-labs/argocd-agent/test/fake/kube"
 	"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 +377,45 @@ func Test_addAppProjectDeletionToQueue(t *testing.T) {
 		require.False(t, a.projectManager.IsManaged("test-project"))
 	})
 }
+
+func Test_addClusterCacheInfoUpdateToQueue(t *testing.T) {
+	miniRedis, err := miniredis.Run()
+	require.NoError(t, err)
+	require.NotNil(t, miniRedis)
+	defer miniRedis.Close()
+
+	kubec := kube.NewKubernetesFakeClientWithApps("argocd")
+	remote, err := client.NewRemote("127.0.0.1", 8080)
+	require.NoError(t, err)
+
+	a, err := NewAgent(context.TODO(), kubec, "argocd", WithRemote(remote), WithRedisHost(miniRedis.Addr()))
+	require.NoError(t, err)
+
+	a.remote.SetClientID("agent")
+	a.emitter = event.NewEventSource("principal")
+
+	// 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())
+}

cmd/argocd-agent/agent.go

@@ -67,6 +67,9 @@ func NewAgentRunCommand() *cobra.Command {
 		// Time interval for agent to principal ping
 		// Ex: "30m", "1h" or "1h20m10s". Valid time units are "s", "m", "h".
 		keepAlivePingInterval time.Duration
+
+		// Time interval for agent to refresh cluster cache info in principal
+		cacheRefreshInterval time.Duration
 	)
 	command := &cobra.Command{
 		Use:   "agent",
@@ -174,6 +177,7 @@ func NewAgentRunCommand() *cobra.Command {
 			agentOpts = append(agentOpts, agent.WithRedisPassword(redisPassword))
 
 			agentOpts = append(agentOpts, agent.WithEnableResourceProxy(enableResourceProxy))
+			agentOpts = append(agentOpts, agent.WithCacheRefreshInterval(cacheRefreshInterval))
 
 			if metricsPort > 0 {
 				agentOpts = append(agentOpts, agent.WithMetricsPort(metricsPort))
@@ -263,6 +267,9 @@ func NewAgentRunCommand() *cobra.Command {
 	command.Flags().BoolVar(&enableResourceProxy, "enable-resource-proxy",
 		env.BoolWithDefault("ARGOCD_AGENT_ENABLE_RESOURCE_PROXY", true),
 		"Enable resource proxy")
+	command.Flags().DurationVar(&cacheRefreshInterval, "cache-refresh-interval",
+		env.DurationWithDefault("ARGOCD_AGENT_CACHE_REFRESH_INTERVAL", nil, 10*time.Second),
+		"Interval to refresh cluster cache info in principal")
 
 	command.Flags().StringVar(&kubeConfig, "kubeconfig", "", "Path to a kubeconfig file to use")
 	command.Flags().StringVar(&kubeContext, "kubecontext", "", "Override the default kube context")