feat: AppProject synchronization

agent/agent.go

@@ -22,18 +22,19 @@ import (
 	"time"
 
 	kubeapp "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/application"
+	kubeappproject "github.com/argoproj-labs/argocd-agent/internal/backend/kubernetes/appproject"
 	"github.com/argoproj-labs/argocd-agent/internal/event"
 	appinformer "github.com/argoproj-labs/argocd-agent/internal/informer/application"
+	appprojectinformer "github.com/argoproj-labs/argocd-agent/internal/informer/appproject"
 	"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/queue"
 	"github.com/argoproj-labs/argocd-agent/internal/version"
 	"github.com/argoproj-labs/argocd-agent/pkg/client"
 	"github.com/argoproj-labs/argocd-agent/pkg/types"
 	"github.com/sirupsen/logrus"
 
-	"k8s.io/client-go/kubernetes"
-
 	appclientset "github.com/argoproj/argo-cd/v2/pkg/client/clientset/versioned"
 )
 
@@ -52,10 +53,11 @@ type Agent struct {
 	infStopCh chan struct{}
 	connected atomic.Bool
 	// syncCh is not currently used
-	syncCh     chan bool
-	remote     *client.Remote
-	appManager *application.ApplicationManager
-	mode       types.AgentMode
+	syncCh         chan bool
+	remote         *client.Remote
+	appManager     *application.ApplicationManager
+	projectManager *appproject.AppProjectManager
+	mode           types.AgentMode
 	// queues is a queue of create/update/delete events to send to the principal
 	queues  *queue.SendRecvQueues
 	emitter *event.EventSource
@@ -79,7 +81,7 @@ type AgentOption func(*Agent) error
 
 // NewAgent creates a new agent instance, using the given client interfaces and
 // options.
-func NewAgent(ctx context.Context, client kubernetes.Interface, appclient appclientset.Interface, namespace string, opts ...AgentOption) (*Agent, error) {
+func NewAgent(ctx context.Context, appclient appclientset.Interface, namespace string, opts ...AgentOption) (*Agent, error) {
 	a := &Agent{
 		version: version.New("argocd-agent", "agent"),
 	}
@@ -118,7 +120,7 @@ func NewAgent(ctx context.Context, client kubernetes.Interface, appclient appcli
 		return nil, fmt.Errorf("unexpected agent mode: %v", a.mode)
 	}
 
-	informer := appinformer.NewAppInformer(ctx, appclient, a.namespace,
+	appInformer := appinformer.NewAppInformer(ctx, appclient, a.namespace,
 		appinformer.WithListAppCallback(a.listAppCallback),
 		appinformer.WithNewAppCallback(a.addAppCreationToQueue),
 		appinformer.WithUpdateAppCallback(a.addAppUpdateToQueue),
@@ -133,9 +135,20 @@ func NewAgent(ctx context.Context, client kubernetes.Interface, appclient appcli
 
 	var err error
 
+	appProjectManagerOption := []appproject.AppProjectManagerOption{
+		appproject.WithAllowUpsert(true),
+		appproject.WithRole(manager.ManagerRoleAgent),
+		appproject.WithMode(managerMode),
+	}
+
+	projectInformer, err := appprojectinformer.NewAppProjectInformer(ctx, appclient, a.namespace)
+	if err != nil {
+		return nil, err
+	}
+
 	// The agent only supports Kubernetes as application backend
 	a.appManager, err = application.NewApplicationManager(
-		kubeapp.NewKubernetesBackend(appclient, a.namespace, informer, true),
+		kubeapp.NewKubernetesBackend(appclient, a.namespace, appInformer, true),
 		a.namespace,
 		application.WithAllowUpsert(allowUpsert),
 		application.WithRole(manager.ManagerRoleAgent),
@@ -146,6 +159,11 @@ func NewAgent(ctx context.Context, client kubernetes.Interface, appclient appcli
 		return nil, err
 	}
 
+	a.projectManager, err = appproject.NewAppProjectManager(kubeappproject.NewKubernetesBackend(appclient, a.namespace, projectInformer, true), a.namespace, appProjectManagerOption...)
+	if err != nil {
+		return nil, err
+	}
+
 	a.syncCh = make(chan bool, 1)
 	return a, nil
 }
@@ -157,7 +175,11 @@ func (a *Agent) Start(ctx context.Context) error {
 	a.cancelFn = cancelFn
 	go func() {
 		a.appManager.StartBackend(a.context)
-		log().Warnf("Informer has exited")
+		log().Warnf("App Informer has exited")
+	}()
+	go func() {
+		a.projectManager.StartBackend(a.context)
+		log().Warnf("Project Informer has exited")
 	}()
 	if a.remote != nil {
 		a.remote.SetClientMode(a.mode)
@@ -168,8 +190,12 @@ func (a *Agent) Start(ctx context.Context) error {
 
 	a.emitter = event.NewEventSource(fmt.Sprintf("agent://%s", "agent-managed"))
 
-	// Wait for the informer to be synced
+	// Wait for the app informer to be synced
 	err := a.appManager.EnsureSynced(waitForSyncedDuration)
+	if err != nil {
+		return fmt.Errorf("failed to sync applications: %w", err)
+	}
+
 	return err
 }
 

agent/agent_test.go

@@ -22,25 +22,22 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/argoproj-labs/argocd-agent/pkg/client"
-	fakekube "github.com/argoproj-labs/argocd-agent/test/fake/kube"
 	"github.com/stretchr/testify/require"
 )
 
 func newAgent(t *testing.T) *Agent {
 	t.Helper()
-	fakec := fakekube.NewFakeClientsetWithResources()
 	appc := fakeappclient.NewSimpleClientset()
 	remote, err := client.NewRemote("127.0.0.1", 8080)
 	require.NoError(t, err)
-	agent, err := NewAgent(context.TODO(), fakec, appc, "argocd", WithRemote(remote))
+	agent, err := NewAgent(context.TODO(), appc, "argocd", WithRemote(remote))
 	require.NoError(t, err)
 	return agent
 }
 
 func Test_NewAgent(t *testing.T) {
-	fakec := fakekube.NewFakeClientsetWithResources()
 	appc := fakeappclient.NewSimpleClientset()
-	agent, err := NewAgent(context.TODO(), fakec, appc, "agent", WithRemote(&client.Remote{}))
+	agent, err := NewAgent(context.TODO(), appc, "agent", WithRemote(&client.Remote{}))
 	require.NotNil(t, agent)
 	require.NoError(t, err)
 }

agent/inbound.go

@@ -37,6 +37,7 @@ func (a *Agent) processIncomingEvent(ev *event.Event) error {
 	case event.TargetApplication:
 		err = a.processIncomingApplication(ev)
 	case event.TargetAppProject:
+		err = a.processIncomingAppProject(ev)
 	default:
 		err = fmt.Errorf("unknown event target: %s", ev.Target())
 	}
@@ -76,6 +77,38 @@ func (a *Agent) processIncomingApplication(ev *event.Event) error {
 	return err
 }
 
+func (a *Agent) processIncomingAppProject(ev *event.Event) error {
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "processIncomingEvents",
+	})
+	incomingAppProject, err := ev.AppProject()
+	if err != nil {
+		return err
+	}
+
+	switch ev.Type() {
+	case event.Create:
+		_, err = a.createAppProject(incomingAppProject)
+		if err != nil {
+			logCtx.Errorf("Error creating appproject: %v", err)
+		}
+	case event.SpecUpdate:
+		_, err = a.updateAppProject(incomingAppProject)
+		if err != nil {
+			logCtx.Errorf("Error updating appproject: %v", err)
+		}
+	case event.Delete:
+		err = a.deleteAppProject(incomingAppProject)
+		if err != nil {
+			logCtx.Errorf("Error deleting appproject: %v", err)
+		}
+	default:
+		logCtx.Warnf("Received an unknown event: %s. Protocol mismatch?", ev.Type())
+	}
+
+	return err
+}
+
 // createApplication creates an Application upon an event in the agent's work
 // queue.
 func (a *Agent) createApplication(incoming *v1alpha1.Application) (*v1alpha1.Application, error) {
@@ -172,3 +205,89 @@ func (a *Agent) deleteApplication(app *v1alpha1.Application) error {
 	}
 	return nil
 }
+
+// createAppProject creates an AppProject upon an event in the agent's work
+// queue.
+func (a *Agent) createAppProject(incoming *v1alpha1.AppProject) (*v1alpha1.AppProject, error) {
+	incoming.ObjectMeta.SetNamespace(a.namespace)
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "CreateAppProject",
+		"app":    incoming.Name,
+	})
+
+	// In modes other than "managed", we don't process new AppProject events
+	// that are incoming.
+	if a.mode != types.AgentModeManaged {
+		logCtx.Trace("Discarding this event, because agent is not in managed mode")
+		return nil, fmt.Errorf("cannot create appproject: agent is not in managed mode")
+	}
+
+	// If we receive a new AppProject event for an AppProject we already manage, it usually
+	// means that we're out-of-sync from the control plane.
+	if a.appManager.IsManaged(incoming.Name) {
+		logCtx.Trace("Discarding this event, because AppProject is already managed on this agent")
+		return nil, fmt.Errorf("appproject %s is already managed", incoming.Name)
+	}
+
+	logCtx.Infof("Creating a new AppProject on behalf of an incoming event")
+
+	// Get rid of some fields that we do not want to have on the AppProject
+	// as we start fresh.
+	if incoming.Annotations != nil {
+		delete(incoming.Annotations, "kubectl.kubernetes.io/last-applied-configuration")
+	}
+
+	created, err := a.projectManager.Create(a.context, incoming)
+	return created, err
+}
+
+func (a *Agent) updateAppProject(incoming *v1alpha1.AppProject) (*v1alpha1.AppProject, error) {
+	//
+	incoming.ObjectMeta.SetNamespace(a.namespace)
+	logCtx := log().WithFields(logrus.Fields{
+		"method":          "UpdateAppProject",
+		"app":             incoming.Name,
+		"resourceVersion": incoming.ResourceVersion,
+	})
+	if a.appManager.IsChangeIgnored(incoming.Name, incoming.ResourceVersion) {
+		logCtx.Tracef("Discarding this event, because agent has seen this version %s already", incoming.ResourceVersion)
+		return nil, fmt.Errorf("the version %s has already been seen by this agent", incoming.ResourceVersion)
+	} else {
+		logCtx.Tracef("New resource version: %s", incoming.ResourceVersion)
+	}
+
+	logCtx.Infof("Updating appProject")
+
+	logCtx.Tracef("Calling update spec for this event")
+	return a.projectManager.UpdateAppProject(a.context, incoming)
+
+}
+
+func (a *Agent) deleteAppProject(project *v1alpha1.AppProject) error {
+	project.ObjectMeta.SetNamespace(a.namespace)
+	logCtx := log().WithFields(logrus.Fields{
+		"method": "DeleteAppProject",
+		"app":    project.Name,
+	})
+
+	// If we receive an update appproject event for an AppProject we don't know about yet it
+	// means that we're out-of-sync from the control plane.
+	//
+	// TODO(jannfis): Handle this situation properly instead of throwing an error.
+	if !a.appManager.IsManaged(project.Name) {
+		return fmt.Errorf("appProject %s is not managed", project.Name)
+	}
+
+	logCtx.Infof("Deleting appProject")
+
+	deletionPropagation := backend.DeletePropagationBackground
+	err := a.projectManager.Delete(a.context, a.namespace, project, &deletionPropagation)
+	if err != nil {
+		return err
+	}
+	err = a.projectManager.Unmanage(project.Name)
+	if err != nil {
+		log().Warnf("Could not unmanage appProject %s: %v", project.Name, err)
+	}
+	return nil
+}

agent/inbound_test.go

@@ -26,6 +26,7 @@ import (
 	backend_mocks "github.com/argoproj-labs/argocd-agent/internal/backend/mocks"
 	"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/pkg/types"
 )
 
@@ -147,3 +148,79 @@ func Test_UpdateApplication(t *testing.T) {
 	})
 
 }
+
+func Test_CreateAppProject(t *testing.T) {
+	a := newAgent(t)
+	be := backend_mocks.NewAppProject(t)
+	var err error
+	a.projectManager, err = appproject.NewAppProjectManager(be, "argocd", appproject.WithAllowUpsert(true))
+	require.NoError(t, err)
+	require.NotNil(t, a)
+	project := &v1alpha1.AppProject{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test",
+			Namespace: "default",
+		}, Spec: v1alpha1.AppProjectSpec{
+			SourceNamespaces: []string{"default", "argocd"},
+		},
+	}
+
+	t.Run("Discard event in unmanaged mode", func(t *testing.T) {
+		napp, err := a.createAppProject(project)
+		require.Nil(t, napp)
+		require.ErrorContains(t, err, "not in managed mode")
+	})
+
+	t.Run("Discard event because appproject is already managed", func(t *testing.T) {
+		defer a.appManager.Unmanage(project.Name)
+		a.mode = types.AgentModeManaged
+		a.appManager.Manage(project.Name)
+		napp, err := a.createAppProject(project)
+		require.ErrorContains(t, err, "is already managed")
+		require.Nil(t, napp)
+	})
+
+	t.Run("Create appproject", func(t *testing.T) {
+		defer a.appManager.Unmanage(project.Name)
+		a.mode = types.AgentModeManaged
+		createMock := be.On("Create", mock.Anything, mock.Anything).Return(&v1alpha1.AppProject{}, nil)
+		defer createMock.Unset()
+		napp, err := a.createAppProject(project)
+		require.NoError(t, err)
+		require.NotNil(t, napp)
+	})
+
+}
+
+func Test_UpdateAppProject(t *testing.T) {
+	a := newAgent(t)
+	be := backend_mocks.NewAppProject(t)
+	var err error
+	a.projectManager, err = appproject.NewAppProjectManager(be, "argocd", appproject.WithAllowUpsert(true))
+	require.NoError(t, err)
+	require.NotNil(t, a)
+	project := &v1alpha1.AppProject{
+		ObjectMeta: v1.ObjectMeta{
+			Name:            "test",
+			Namespace:       "argocd",
+			ResourceVersion: "12345",
+		},
+		Spec: v1alpha1.AppProjectSpec{
+			SourceNamespaces: []string{"default", "argocd"},
+		}}
+
+	t.Run("Update appproject using patch", func(t *testing.T) {
+		a.mode = types.AgentModeManaged
+		a.projectManager, err = appproject.NewAppProjectManager(be, "argocd", appproject.WithAllowUpsert(true), appproject.WithMode(manager.ManagerModeManaged), appproject.WithRole(manager.ManagerRoleAgent))
+		getEvent := be.On("Get", mock.Anything, mock.Anything, mock.Anything).Return(&v1alpha1.AppProject{}, nil)
+		defer getEvent.Unset()
+		supportsPatchEvent := be.On("SupportsPatch").Return(true)
+		defer supportsPatchEvent.Unset()
+		patchEvent := be.On("Patch", mock.Anything, "test", "argocd", mock.Anything).Return(&v1alpha1.AppProject{}, nil)
+		defer patchEvent.Unset()
+		napp, err := a.updateAppProject(project)
+		require.NoError(t, err)
+		require.NotNil(t, napp)
+	})
+
+}

cmd/agent/main.go

@@ -117,7 +117,7 @@ func NewAgentRunCommand() *cobra.Command {
 			}
 			agentOpts = append(agentOpts, agent.WithRemote(remote))
 			agentOpts = append(agentOpts, agent.WithMode(agentMode))
-			ag, err := agent.NewAgent(ctx, kubeConfig.Clientset, kubeConfig.ApplicationsClientset, namespace, agentOpts...)
+			ag, err := agent.NewAgent(ctx, kubeConfig.ApplicationsClientset, namespace, agentOpts...)
 			if err != nil {
 				cmd.Fatal("Could not create a new agent instance: %v", err)
 			}

hack/demo-env/control-plane/kustomization.yaml

@@ -15,3 +15,4 @@ patches:
 - path: server-service.yaml
 - path: repo-server-service.yaml
 - path: redis-service.yaml
+- path: appproject-default.yaml

hack/demo-env/gen-creds.sh

@@ -42,4 +42,4 @@ for ag in agent-managed agent-autonomous; do
 	password=$(pwmake 56)
 	htpasswd -b -B "${creds_path}/users.control-plane" "${ag}" "${password}"
 	echo "${ag}:${password}" > "${creds_path}/creds.${ag}"
-done
+done

hack/demo-env/resources/metallb-ipaddresspool.yaml

@@ -3,7 +3,6 @@ kind: IPAddressPool
 metadata:
   name: default-addresspool
   namespace: metallb-system
-  uid: 1812d024-a3f6-4c4e-8aa3-68d2260618ed
 spec:
   addresses:
   - 192.168.56.100-192.168.56.254