Merge branch 'argoproj-labs:main' into main

Author: juanxiu

agent/agent.go

@@ -22,17 +22,20 @@ 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"
 	kubenamespace "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/namespace"
+	kuberepository "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/repository"
 	"github.com/argoproj-labs/argocd-agent/internal/event"
 	"github.com/argoproj-labs/argocd-agent/internal/informer"
 	"github.com/argoproj-labs/argocd-agent/internal/kube"
 	"github.com/argoproj-labs/argocd-agent/internal/logging"
 	"github.com/argoproj-labs/argocd-agent/internal/manager"
 	"github.com/argoproj-labs/argocd-agent/internal/manager/application"
 	"github.com/argoproj-labs/argocd-agent/internal/manager/appproject"
+	"github.com/argoproj-labs/argocd-agent/internal/manager/repository"
 	"github.com/argoproj-labs/argocd-agent/internal/metrics"
 	"github.com/argoproj-labs/argocd-agent/internal/queue"
 	"github.com/argoproj-labs/argocd-agent/internal/resources"
@@ -47,6 +50,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"
 )
 
@@ -69,6 +74,7 @@ type Agent struct {
 	remote           *client.Remote
 	appManager       *application.ApplicationManager
 	projectManager   *appproject.AppProjectManager
+	repoManager      *repository.RepositoryManager
 	namespaceManager *kubenamespace.KubernetesBackend
 	mode             types.AgentMode
 	// queues is a queue of create/update/delete events to send to the principal
@@ -94,6 +100,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"
@@ -242,6 +251,27 @@ func NewAgent(ctx context.Context, client *kube.KubernetesClient, namespace stri
 		return nil, err
 	}
 
+	repoInformerOptions := []informer.InformerOption[*corev1.Secret]{
+		informer.WithListHandler[*corev1.Secret](func(ctx context.Context, opts v1.ListOptions) (runtime.Object, error) {
+			return client.Clientset.CoreV1().Secrets(a.namespace).List(ctx, opts)
+		}),
+		informer.WithWatchHandler[*corev1.Secret](func(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+			return client.Clientset.CoreV1().Secrets(a.namespace).Watch(ctx, opts)
+		}),
+		informer.WithAddHandler[*corev1.Secret](a.handleRepositoryCreation),
+		informer.WithUpdateHandler[*corev1.Secret](a.handleRepositoryUpdate),
+		informer.WithDeleteHandler[*corev1.Secret](a.handleRepositoryDeletion),
+		informer.WithFilters(kuberepository.DefaultFilterChain(a.namespace)),
+	}
+
+	repoInformer, err := informer.NewInformer(ctx, repoInformerOptions...)
+	if err != nil {
+		return nil, fmt.Errorf("could not instantiate repository informer: %w", err)
+	}
+
+	repoBackened := kuberepository.NewKubernetesBackend(client.Clientset, a.namespace, repoInformer, true)
+	a.repoManager = repository.NewManager(repoBackened, a.namespace, true)
+
 	nsInformerOpts := []informer.InformerOption[*corev1.Namespace]{
 		informer.WithListHandler[*corev1.Namespace](func(ctx context.Context, opts v1.ListOptions) (runtime.Object, error) {
 			return client.Clientset.CoreV1().Namespaces().List(ctx, opts)
@@ -273,6 +303,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
 }
 
@@ -329,6 +366,12 @@ func (a *Agent) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to sync applications: %w", err)
 	}
 
+	// Wait for the appProject informer to be synced
+	err = a.projectManager.EnsureSynced(waitForSyncedDuration)
+	if err != nil {
+		return fmt.Errorf("failed to sync appProjects: %w", err)
+	}
+
 	// The namespace informer lives in its own go routine
 	go func() {
 		if err := a.namespaceManager.StartInformer(a.context); err != nil {
@@ -343,6 +386,20 @@ func (a *Agent) Start(ctx context.Context) error {
 	}
 	log().Infof("Namespace informer synced and ready")
 
+	// Start the Repository backend in the background
+	go func() {
+		if err := a.repoManager.StartBackend(a.context); err != nil {
+			log().WithError(err).Error("Repository backend has exited non-successfully")
+		} else {
+			log().Info("Repository backend has exited")
+		}
+	}()
+
+	if err = a.repoManager.EnsureSynced(waitForSyncedDuration); err != nil {
+		return fmt.Errorf("unable to sync Repository informer: %w", err)
+	}
+	log().Infof("Repository informer synced and ready")
+
 	if a.options.healthzPort > 0 {
 		// Endpoint to check if the agent is up and running
 		http.HandleFunc("/healthz", a.healthzHandler)
@@ -353,6 +410,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

@@ -28,6 +28,7 @@ import (
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
 	"github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/dynamic"
 )
@@ -57,6 +58,8 @@ func (a *Agent) processIncomingEvent(ev *event.Event) error {
 		err = a.processIncomingApplication(ev)
 	case event.TargetAppProject:
 		err = a.processIncomingAppProject(ev)
+	case event.TargetRepository:
+		err = a.processIncomingRepository(ev)
 	case event.TargetResource:
 		err = a.processIncomingResourceRequest(ev)
 	case event.TargetResourceResync:
@@ -264,6 +267,88 @@ func (a *Agent) processIncomingAppProject(ev *event.Event) error {
 	return err
 }
 
+func (a *Agent) processIncomingRepository(ev *event.Event) error {
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "processIncomingEvents",
+	})
+
+	incomingRepo, err := ev.Repository()
+	if err != nil {
+		return err
+	}
+
+	var exists, sourceUIDMatch bool
+
+	// Source UID annotation is not present for repos on the autonomous agent since it is the source of truth.
+	if a.mode == types.AgentModeManaged {
+		exists, sourceUIDMatch, err = a.repoManager.CompareSourceUID(a.context, incomingRepo)
+		if err != nil {
+			return fmt.Errorf("failed to compare the source UID of app: %w", err)
+		}
+	}
+
+	switch ev.Type() {
+	case event.Create:
+		if exists {
+			if sourceUIDMatch {
+				logCtx.Debug("Received a Create event for an existing repository. Updating the existing repository")
+				_, err := a.updateRepository(incomingRepo)
+				if err != nil {
+					return fmt.Errorf("could not update the existing repository: %w", err)
+				}
+				return nil
+			} else {
+				logCtx.Debug("Repository already exists with a different source UID. Deleting the existing repository")
+				if err := a.deleteRepository(incomingRepo); err != nil {
+					return fmt.Errorf("could not delete existing repository prior to creation: %w", err)
+				}
+			}
+		}
+
+		_, err = a.createRepository(incomingRepo)
+		if err != nil {
+			logCtx.Errorf("Error creating repository: %v", err)
+		}
+
+	case event.SpecUpdate:
+		if !exists {
+			logCtx.Debug("Received an Update event for a repository that doesn't exist. Creating the incoming repository")
+			if _, err := a.createRepository(incomingRepo); err != nil {
+				return fmt.Errorf("could not create incoming repository: %w", err)
+			}
+			return nil
+		}
+
+		if !sourceUIDMatch {
+			logCtx.Debug("Source UID mismatch between the incoming repository and existing repository. Deleting the existing repository")
+			if err := a.deleteRepository(incomingRepo); err != nil {
+				return fmt.Errorf("could not delete existing repository prior to creation: %w", err)
+			}
+
+			logCtx.Debug("Creating the incoming repository after deleting the existing repository")
+			if _, err := a.createRepository(incomingRepo); err != nil {
+				return fmt.Errorf("could not create incoming repository after deleting existing repository: %w", err)
+			}
+			return nil
+		}
+
+		_, err = a.updateRepository(incomingRepo)
+		if err != nil {
+			logCtx.Errorf("Error updating repository: %v", err)
+		}
+
+	case event.Delete:
+		err = a.deleteRepository(incomingRepo)
+		if err != nil {
+			logCtx.Errorf("Error deleting repository: %v", err)
+		}
+	default:
+		logCtx.Warnf("Received an unknown event: %s. Protocol mismatch?", ev.Type())
+	}
+
+	return err
+}
+
 // processIncomingResourceResyncEvent handles all the resync events that are
 // exchanged with the agent/principal restarts
 func (a *Agent) processIncomingResourceResyncEvent(ev *event.Event) error {
@@ -457,7 +542,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)
 	}
 
@@ -562,3 +647,98 @@ func (a *Agent) deleteAppProject(project *v1alpha1.AppProject) error {
 	}
 	return nil
 }
+
+// createRepository creates a Repository upon an event in the agent's work queue.
+func (a *Agent) createRepository(incoming *corev1.Secret) (*corev1.Secret, error) {
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "CreateRepository",
+		"repo":   incoming.Name,
+	})
+
+	// In modes other than "managed", we don't process new repository events
+	// that are incoming.
+	if a.mode.IsAutonomous() {
+		logCtx.Info("Discarding this event, because agent is not in managed mode")
+		return nil, event.NewEventDiscardedErr("cannot create repository: agent is not in managed mode")
+	}
+
+	// If we receive a new Repository event for a Repository we already manage, it usually
+	// means that we're out-of-sync from the control plane.
+	if a.repoManager.IsManaged(incoming.Name) {
+		logCtx.Trace("Repository is already managed on this agent. Updating the existing Repository")
+		return a.updateRepository(incoming)
+	}
+
+	logCtx.Infof("Creating a new repository on behalf of an incoming event")
+
+	if incoming.Annotations == nil {
+		incoming.Annotations = make(map[string]string)
+	}
+
+	// Get rid of some fields that we do not want to have on the repository as we start fresh.
+	delete(incoming.Annotations, "kubectl.kubernetes.io/last-applied-configuration")
+
+	created, err := a.repoManager.Create(a.context, incoming)
+	if apierrors.IsAlreadyExists(err) {
+		logCtx.Debug("repository already exists")
+		return created, nil
+	}
+
+	return created, err
+}
+
+func (a *Agent) updateRepository(incoming *corev1.Secret) (*corev1.Secret, error) {
+	incoming.SetNamespace(a.namespace)
+	logCtx := log().WithFields(logrus.Fields{
+		"method":          "UpdateRepository",
+		"repo":            incoming.Name,
+		"resourceVersion": incoming.ResourceVersion,
+	})
+
+	if !a.repoManager.IsManaged(incoming.Name) {
+		logCtx.Trace("Repository is not managed on this agent. Creating the new Repository")
+		return a.createRepository(incoming)
+	}
+
+	if a.repoManager.IsChangeIgnored(incoming.Name, incoming.ResourceVersion) {
+		logCtx.Tracef("Discarding this event, because agent has seen this version %s already", incoming.ResourceVersion)
+		return nil, event.NewEventDiscardedErr("the version %s has already been seen by this agent", incoming.ResourceVersion)
+	} else {
+		logCtx.Tracef("New resource version: %s", incoming.ResourceVersion)
+	}
+
+	logCtx.Infof("Updating repository")
+
+	return a.repoManager.UpdateManagedRepository(a.context, incoming)
+}
+
+func (a *Agent) deleteRepository(repo *corev1.Secret) error {
+	repo.SetNamespace(a.namespace)
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "DeleteRepository",
+		"repo":   repo.Name,
+	})
+
+	if !a.repoManager.IsManaged(repo.Name) {
+		return fmt.Errorf("repository %s is not managed", repo.Name)
+	}
+
+	logCtx.Infof("Deleting repository")
+
+	deletionPropagation := backend.DeletePropagationBackground
+	err := a.repoManager.Delete(a.context, repo.Name, repo.Namespace, &deletionPropagation)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			logCtx.Debug("repository is not found, perhaps it is already deleted")
+			return nil
+		}
+		return err
+	}
+
+	err = a.repoManager.Unmanage(repo.Name)
+	if err != nil {
+		log().Warnf("Could not unmanage repository %s: %v", repo.Name, err)
+	}
+
+	return nil
+}