feat: ApplicationSet watch API (#26409)

Signed-off-by: nitishfy <justnitish06@gmail.com>
Signed-off-by: Peter Jiang <peterjiang823@gmail.com>
Co-authored-by: nitishfy <justnitish06@gmail.com>

.mockery.yaml

@@ -46,7 +46,7 @@ packages:
     interfaces:
       RepoServerServiceClient: {}
       RepoServerService_GenerateManifestWithFilesClient: {}
-  github.com/argoproj/argo-cd/v3/server/application:
+  github.com/argoproj/argo-cd/v3/server/broadcast:
     interfaces:
       Broadcaster: {}
   github.com/argoproj/argo-cd/v3/server/extension:

assets/swagger.json

@@ -4385,6 +4385,69 @@
         }
       }
     },
+    "/api/v1/stream/applicationsets": {
+      "get": {
+        "tags": [
+          "ApplicationSetService"
+        ],
+        "operationId": "ApplicationSetService_Watch",
+        "parameters": [
+          {
+            "type": "string",
+            "name": "name",
+            "in": "query"
+          },
+          {
+            "type": "array",
+            "items": {
+              "type": "string"
+            },
+            "collectionFormat": "multi",
+            "name": "projects",
+            "in": "query"
+          },
+          {
+            "type": "string",
+            "name": "selector",
+            "in": "query"
+          },
+          {
+            "type": "string",
+            "name": "appSetNamespace",
+            "in": "query"
+          },
+          {
+            "type": "string",
+            "description": "when specified with a watch call, shows changes that occur after that particular version of a resource.",
+            "name": "resourceVersion",
+            "in": "query"
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "A successful response.(streaming responses)",
+            "schema": {
+              "type": "object",
+              "title": "Stream result of v1alpha1ApplicationSetWatchEvent",
+              "properties": {
+                "error": {
+                  "$ref": "#/definitions/runtimeStreamError"
+                },
+                "result": {
+                  "$ref": "#/definitions/v1alpha1ApplicationSetWatchEvent"
+                }
+              }
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/runtimeError"
+            }
+          }
+        }
+      }
+    },
     "/api/v1/write-repocreds": {
       "get": {
         "tags": [
@@ -7459,6 +7522,19 @@
         }
       }
     },
+    "v1alpha1ApplicationSetWatchEvent": {
+      "description": "ApplicationSetWatchEvent contains information about application change.",
+      "type": "object",
+      "properties": {
+        "applicationSet": {
+          "$ref": "#/definitions/v1alpha1ApplicationSet"
+        },
+        "type": {
+          "type": "string",
+          "title": "Type represents the Kubernetes watch event type. The protobuf tag uses\ncasttype to ensure the generated Go code keeps this field as\nwatch.EventType (a strong Go type) instead of falling back to a plain string"
+        }
+      }
+    },
     "v1alpha1ApplicationSource": {
       "type": "object",
       "title": "ApplicationSource contains all required information about the source of an application",

cmd/argocd/commands/app_test.go

@@ -2386,3 +2386,22 @@ func (c *fakeAcdClient) WatchApplicationWithRetry(_ context.Context, _ string, _
 	}()
 	return appEventsCh
 }
+
+func (c *fakeAcdClient) WatchApplicationSetWithRetry(_ context.Context, _ string, _ string) chan *v1alpha1.ApplicationSetWatchEvent {
+	appSetEventsCh := make(chan *v1alpha1.ApplicationSetWatchEvent)
+	go func() {
+		defer close(appSetEventsCh)
+		addedEvent := &v1alpha1.ApplicationSetWatchEvent{
+			Type: watch.Added,
+			ApplicationSet: v1alpha1.ApplicationSet{
+				Status: v1alpha1.ApplicationSetStatus{
+					Conditions: []v1alpha1.ApplicationSetCondition{
+						{Type: v1alpha1.ApplicationSetConditionResourcesUpToDate, Status: v1alpha1.ApplicationSetConditionStatusTrue},
+					},
+				},
+			},
+		}
+		appSetEventsCh <- addedEvent
+	}()
+	return appSetEventsCh
+}

cmd/argocd/commands/applicationset.go

@@ -1,12 +1,15 @@
 package commands
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"os"
 	"reflect"
 	"text/tabwriter"
 
+	k8swatch "k8s.io/apimachinery/pkg/watch"
+
 	"github.com/mattn/go-isatty"
 	"github.com/spf13/cobra"
 	"google.golang.org/grpc/codes"
@@ -115,8 +118,10 @@ func NewApplicationSetGetCommand(clientOpts *argocdclient.ClientOptions) *cobra.
 
 // NewApplicationSetCreateCommand returns a new instance of an `argocd appset create` command
 func NewApplicationSetCreateCommand(clientOpts *argocdclient.ClientOptions) *cobra.Command {
-	var output string
-	var upsert, dryRun bool
+	var (
+		output               string
+		upsert, dryRun, wait bool
+	)
 	command := &cobra.Command{
 		Use:   "create",
 		Short: "Create one or more ApplicationSets",
@@ -200,11 +205,18 @@ func NewApplicationSetCreateCommand(clientOpts *argocdclient.ClientOptions) *cob
 				default:
 					errors.CheckError(fmt.Errorf("unknown output format: %s", output))
 				}
+
+				if wait && !dryRun {
+					err := waitForApplicationSetResourcesUpToDate(ctx, argocdClient, created.QualifiedName())
+					errors.CheckError(err)
+					c.PrintErrf("ApplicationSet '%s' resources are up to date\n", created.Name)
+				}
 			}
 		},
 	}
 	command.Flags().BoolVar(&upsert, "upsert", false, "Allows to override ApplicationSet with the same name even if supplied ApplicationSet spec is different from existing spec")
 	command.Flags().BoolVar(&dryRun, "dry-run", false, "Allows to evaluate the ApplicationSet template on the server to get a preview of the applications that would be created")
+	command.Flags().BoolVar(&wait, "wait", false, "Wait until the ApplicationSet's resources are up to date. Will block indefinitely if the ApplicationSet has errors")
 	command.Flags().StringVarP(&output, "output", "o", "wide", "Output format. One of: json|yaml|wide")
 	return command
 }
@@ -325,7 +337,10 @@ func NewApplicationSetListCommand(clientOpts *argocdclient.ClientOptions) *cobra
 
 // NewApplicationSetDeleteCommand returns a new instance of an `argocd appset delete` command
 func NewApplicationSetDeleteCommand(clientOpts *argocdclient.ClientOptions) *cobra.Command {
-	var noPrompt bool
+	var (
+		noPrompt bool
+		wait     bool
+	)
 	command := &cobra.Command{
 		Use:   "delete",
 		Short: "Delete one or more ApplicationSets",
@@ -340,7 +355,8 @@ func NewApplicationSetDeleteCommand(clientOpts *argocdclient.ClientOptions) *cob
 				c.HelpFunc()(c, args)
 				os.Exit(1)
 			}
-			conn, appIf := headless.NewClientOrDie(clientOpts, c).NewApplicationSetClientOrDie()
+			acdClient := headless.NewClientOrDie(clientOpts, c)
+			conn, appIf := acdClient.NewApplicationSetClientOrDie()
 			defer utilio.Close(conn)
 			isTerminal := isatty.IsTerminal(os.Stdout.Fd()) || isatty.IsCygwinTerminal(os.Stdout.Fd())
 			numOfApps := len(args)
@@ -373,6 +389,20 @@ func NewApplicationSetDeleteCommand(clientOpts *argocdclient.ClientOptions) *cob
 				if confirm || confirmAll {
 					_, err := appIf.Delete(ctx, &appsetDeleteReq)
 					errors.CheckError(err)
+
+					if wait {
+						_, getErr := appIf.Get(ctx, &applicationset.ApplicationSetGetQuery{
+							Name: appSetName, AppsetNamespace: appSetNs,
+						})
+						if getErr == nil {
+							appEventCh := acdClient.WatchApplicationSetWithRetry(ctx, appSetQualifiedName, "")
+							for appEvent := range appEventCh {
+								if appEvent != nil && appEvent.Type == k8swatch.Deleted {
+									break
+								}
+							}
+						}
+					}
 					fmt.Printf("applicationset '%s' deleted\n", appSetQualifiedName)
 				} else {
 					fmt.Println("The command to delete '" + appSetQualifiedName + "' was cancelled.")
@@ -381,6 +411,7 @@ func NewApplicationSetDeleteCommand(clientOpts *argocdclient.ClientOptions) *cob
 		},
 	}
 	command.Flags().BoolVarP(&noPrompt, "yes", "y", false, "Turn off prompting to confirm cascaded deletion of Application resources")
+	command.Flags().BoolVar(&wait, "wait", false, "Wait until deletion of the applicationset(s) completes")
 	return command
 }
 
@@ -478,6 +509,31 @@ func printAppSetConditions(w io.Writer, appSet *arogappsetv1.ApplicationSet) {
 	}
 }
 
+func isApplicationSetResourcesUpToDate(appSet *arogappsetv1.ApplicationSet) bool {
+	for _, c := range appSet.Status.Conditions {
+		if c.Type == arogappsetv1.ApplicationSetConditionResourcesUpToDate {
+			return c.Status == arogappsetv1.ApplicationSetConditionStatusTrue
+		}
+	}
+	return false
+}
+
+func waitForApplicationSetResourcesUpToDate(ctx context.Context, acdClient argocdclient.Client, appSetName string) error {
+	appEventCh := acdClient.WatchApplicationSetWithRetry(ctx, appSetName, "")
+	for appEvent := range appEventCh {
+		if appEvent == nil {
+			continue
+		}
+		if appEvent.Type == k8swatch.Deleted {
+			return fmt.Errorf("ApplicationSet %q was deleted before reaching ResourcesUpToDate", appSetName)
+		}
+		if isApplicationSetResourcesUpToDate(&appEvent.ApplicationSet) {
+			return nil
+		}
+	}
+	return fmt.Errorf("watch stream closed for ApplicationSet %q before reaching ResourcesUpToDate", appSetName)
+}
+
 func hasAppSetChanged(appReq, appRes *arogappsetv1.ApplicationSet, upsert bool) bool {
 	// upsert==false, no change occurred from create command
 	if !upsert {

cmd/argocd/commands/applicationset_test.go

@@ -1,6 +1,8 @@
 package commands
 
 import (
+	"context"
+	"errors"
 	"io"
 	"os"
 	"testing"
@@ -8,10 +10,94 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
 
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
 )
 
+// TestAppSetDeleteWaitFlow verifies that when --wait is used and the appset has
+// finalizers (still exists after Delete), the delete command watches for Deleted.
+func TestAppSetDeleteWaitFlow(t *testing.T) {
+	appSetEventsCh := make(chan *v1alpha1.ApplicationSetWatchEvent, 1)
+	go func() {
+		defer close(appSetEventsCh)
+		appSetEventsCh <- &v1alpha1.ApplicationSetWatchEvent{
+			Type:           watch.Added,
+			ApplicationSet: v1alpha1.ApplicationSet{ObjectMeta: metav1.ObjectMeta{Name: "test-appset"}},
+		}
+		appSetEventsCh <- &v1alpha1.ApplicationSetWatchEvent{Type: watch.Deleted}
+	}()
+
+	receivedDeleted := false
+	for appEvent := range appSetEventsCh {
+		if appEvent != nil && appEvent.Type == watch.Deleted {
+			receivedDeleted = true
+			break
+		}
+	}
+	assert.True(t, receivedDeleted, "wait loop should receive Deleted event from watch")
+}
+
+// TestAppSetCreateWaitFlow verifies that when --wait is used, the create command
+// waits for ResourcesUpToDate from the watch before completing.
+func TestAppSetCreateWaitFlow(t *testing.T) {
+	fakeClient := &fakeAcdClient{}
+	ctx := context.Background()
+
+	err := waitForApplicationSetResourcesUpToDate(ctx, fakeClient, "test-appset")
+	require.NoError(t, err)
+}
+
+func TestAppSetCreateWaitDeletedError(t *testing.T) {
+	appSetEventsCh := make(chan *v1alpha1.ApplicationSetWatchEvent, 1)
+	go func() {
+		defer close(appSetEventsCh)
+		appSetEventsCh <- &v1alpha1.ApplicationSetWatchEvent{Type: watch.Deleted}
+	}()
+
+	var err error
+	for appEvent := range appSetEventsCh {
+		if appEvent == nil {
+			continue
+		}
+		if appEvent.Type == watch.Deleted {
+			err = errors.New("ApplicationSet was deleted before reaching ResourcesUpToDate")
+			break
+		}
+	}
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "deleted before reaching ResourcesUpToDate")
+}
+
+func TestIsApplicationSetResourcesUpToDate(t *testing.T) {
+	t.Run("returns true when ResourcesUpToDate is True", func(t *testing.T) {
+		appSet := &v1alpha1.ApplicationSet{
+			Status: v1alpha1.ApplicationSetStatus{
+				Conditions: []v1alpha1.ApplicationSetCondition{
+					{Type: v1alpha1.ApplicationSetConditionResourcesUpToDate, Status: v1alpha1.ApplicationSetConditionStatusTrue},
+				},
+			},
+		}
+		assert.True(t, isApplicationSetResourcesUpToDate(appSet))
+	})
+
+	t.Run("returns false when ResourcesUpToDate is False", func(t *testing.T) {
+		appSet := &v1alpha1.ApplicationSet{
+			Status: v1alpha1.ApplicationSetStatus{
+				Conditions: []v1alpha1.ApplicationSetCondition{
+					{Type: v1alpha1.ApplicationSetConditionResourcesUpToDate, Status: v1alpha1.ApplicationSetConditionStatusFalse},
+				},
+			},
+		}
+		assert.False(t, isApplicationSetResourcesUpToDate(appSet))
+	})
+
+	t.Run("returns false when no conditions", func(t *testing.T) {
+		appSet := &v1alpha1.ApplicationSet{}
+		assert.False(t, isApplicationSetResourcesUpToDate(appSet))
+	})
+}
+
 func TestPrintApplicationSetNames(t *testing.T) {
 	output, _ := captureOutput(func() error {
 		appSet := &v1alpha1.ApplicationSet{

common/common.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -15,6 +16,8 @@ import (
 	"google.golang.org/grpc/status"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+
+	"github.com/argoproj/argo-cd/v3/util/env"
 )
 
 // Argo CD component names
@@ -474,6 +477,8 @@ var ErrTokenVerification = errors.New(TokenVerificationError)
 
 var PermissionDeniedAPIError = status.Error(codes.PermissionDenied, "permission denied")
 
+var WatchAPIBufferSize = env.ParseNumFromEnv(EnvWatchAPIBufferSize, 1000, 0, math.MaxInt32)
+
 // Redis password consts
 const (
 	// RedisInitialCredentials is the name for the argocd kubernetes secret which will have the redis password

docs/user-guide/commands/argocd_appset_create.md

@@ -25,6 +25,7 @@ argocd appset create [flags]
   -h, --help            help for create
   -o, --output string   Output format. One of: json|yaml|wide (default "wide")
       --upsert          Allows to override ApplicationSet with the same name even if supplied ApplicationSet spec is different from existing spec
+      --wait            Wait until the ApplicationSet's resources are up to date. Will block indefinitely if the ApplicationSet has errors
 ```
 
 ### Options inherited from parent commands

docs/user-guide/commands/argocd_appset_delete.md

@@ -19,6 +19,7 @@ argocd appset delete [flags]
 
 ```
   -h, --help   help for delete
+      --wait   Wait until deletion of the applicationset(s) completes
   -y, --yes    Turn off prompting to confirm cascaded deletion of Application resources
 ```
 

pkg/apiclient/apiclient.go

@@ -99,6 +99,7 @@ type Client interface {
 	NewAccountClient() (io.Closer, accountpkg.AccountServiceClient, error)
 	NewAccountClientOrDie() (io.Closer, accountpkg.AccountServiceClient)
 	WatchApplicationWithRetry(ctx context.Context, appName string, revision string) chan *v1alpha1.ApplicationWatchEvent
+	WatchApplicationSetWithRetry(ctx context.Context, appSetName, revision string) chan *v1alpha1.ApplicationSetWatchEvent
 }
 
 // ClientOptions hold address, security, and other settings for the API client.
@@ -802,6 +803,47 @@ func (c *client) NewAccountClientOrDie() (io.Closer, accountpkg.AccountServiceCl
 	return conn, usrIf
 }
 
+func (c *client) WatchApplicationSetWithRetry(ctx context.Context, appSetName, _ string) chan *v1alpha1.ApplicationSetWatchEvent {
+	appSetEventCh := make(chan *v1alpha1.ApplicationSetWatchEvent)
+	cancelled := false
+	appSetName, appSetNs := argo.ParseFromQualifiedName(appSetName, "")
+	go func() {
+		defer close(appSetEventCh)
+		for !cancelled {
+			conn, appsetIf, err := c.NewApplicationSetClient()
+			if err == nil {
+				var wc applicationsetpkg.ApplicationSetService_WatchClient
+				wc, err = appsetIf.Watch(ctx, &applicationsetpkg.ApplicationSetWatchQuery{
+					Name:            appSetName,
+					AppSetNamespace: appSetNs,
+				})
+				if err == nil {
+					for {
+						var appSetEvent *v1alpha1.ApplicationSetWatchEvent
+						appSetEvent, err = wc.Recv()
+						if err != nil {
+							break
+						}
+						appSetEventCh <- appSetEvent
+					}
+				}
+			}
+			if err != nil {
+				if isCanceledContextErr(err) {
+					cancelled = true
+				} else {
+					time.Sleep(1 * time.Second)
+				}
+			}
+			if conn != nil {
+				_ = conn.Close()
+			}
+		}
+	}()
+
+	return appSetEventCh
+}
+
 // WatchApplicationWithRetry returns a channel of watch events for an application, retrying the
 // watch upon errors. Closes the returned channel when the context is cancelled.
 func (c *client) WatchApplicationWithRetry(ctx context.Context, appName string, revision string) chan *v1alpha1.ApplicationWatchEvent {

pkg/apiclient/applicationset/applicationset.pb.go

@@ -157,6 +157,86 @@ func (m *ApplicationSetListQuery) GetAppsetNamespace() string {
 	return ""
 }
 
+type ApplicationSetWatchQuery struct {
+	Name            string   `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	Projects        []string `protobuf:"bytes,2,rep,name=projects,proto3" json:"projects,omitempty"`
+	Selector        string   `protobuf:"bytes,3,opt,name=selector,proto3" json:"selector,omitempty"`
+	AppSetNamespace string   `protobuf:"bytes,4,opt,name=appSetNamespace,proto3" json:"appSetNamespace,omitempty"`
+	// when specified with a watch call, shows changes that occur after that particular version of a resource.
+	ResourceVersion      string   `protobuf:"bytes,5,opt,name=resourceVersion,proto3" json:"resourceVersion,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ApplicationSetWatchQuery) Reset()         { *m = ApplicationSetWatchQuery{} }
+func (m *ApplicationSetWatchQuery) String() string { return proto.CompactTextString(m) }
+func (*ApplicationSetWatchQuery) ProtoMessage()    {}
+func (*ApplicationSetWatchQuery) Descriptor() ([]byte, []int) {
+	return fileDescriptor_eacb9df0ce5738fa, []int{2}
+}
+func (m *ApplicationSetWatchQuery) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *ApplicationSetWatchQuery) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_ApplicationSetWatchQuery.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *ApplicationSetWatchQuery) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ApplicationSetWatchQuery.Merge(m, src)
+}
+func (m *ApplicationSetWatchQuery) XXX_Size() int {
+	return m.Size()
+}
+func (m *ApplicationSetWatchQuery) XXX_DiscardUnknown() {
+	xxx_messageInfo_ApplicationSetWatchQuery.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ApplicationSetWatchQuery proto.InternalMessageInfo
+
+func (m *ApplicationSetWatchQuery) GetName() string {
+	if m != nil {
+		return m.Name
+	}
+	return ""
+}
+
+func (m *ApplicationSetWatchQuery) GetProjects() []string {
+	if m != nil {
+		return m.Projects
+	}
+	return nil
+}
+
+func (m *ApplicationSetWatchQuery) GetSelector() string {
+	if m != nil {
+		return m.Selector
+	}
+	return ""
+}
+
+func (m *ApplicationSetWatchQuery) GetAppSetNamespace() string {
+	if m != nil {
+		return m.AppSetNamespace
+	}
+	return ""
+}
+
+func (m *ApplicationSetWatchQuery) GetResourceVersion() string {
+	if m != nil {
+		return m.ResourceVersion
+	}
+	return ""
+}
+
 type ApplicationSetResponse struct {
 	Project              string                   `protobuf:"bytes,1,opt,name=project,proto3" json:"project,omitempty"`
 	Applicationset       *v1alpha1.ApplicationSet `protobuf:"bytes,2,opt,name=applicationset,proto3" json:"applicationset,omitempty"`
@@ -169,7 +249,7 @@ func (m *ApplicationSetResponse) Reset()         { *m = ApplicationSetResponse{}
 func (m *ApplicationSetResponse) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetResponse) ProtoMessage()    {}
 func (*ApplicationSetResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{2}
+	return fileDescriptor_eacb9df0ce5738fa, []int{3}
 }
 func (m *ApplicationSetResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -225,7 +305,7 @@ func (m *ApplicationSetCreateRequest) Reset()         { *m = ApplicationSetCreat
 func (m *ApplicationSetCreateRequest) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetCreateRequest) ProtoMessage()    {}
 func (*ApplicationSetCreateRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{3}
+	return fileDescriptor_eacb9df0ce5738fa, []int{4}
 }
 func (m *ApplicationSetCreateRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -288,7 +368,7 @@ func (m *ApplicationSetDeleteRequest) Reset()         { *m = ApplicationSetDelet
 func (m *ApplicationSetDeleteRequest) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetDeleteRequest) ProtoMessage()    {}
 func (*ApplicationSetDeleteRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{4}
+	return fileDescriptor_eacb9df0ce5738fa, []int{5}
 }
 func (m *ApplicationSetDeleteRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -344,7 +424,7 @@ func (m *ApplicationSetTreeQuery) Reset()         { *m = ApplicationSetTreeQuery
 func (m *ApplicationSetTreeQuery) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetTreeQuery) ProtoMessage()    {}
 func (*ApplicationSetTreeQuery) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{5}
+	return fileDescriptor_eacb9df0ce5738fa, []int{6}
 }
 func (m *ApplicationSetTreeQuery) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -400,7 +480,7 @@ func (m *ApplicationSetGenerateRequest) Reset()         { *m = ApplicationSetGen
 func (m *ApplicationSetGenerateRequest) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetGenerateRequest) ProtoMessage()    {}
 func (*ApplicationSetGenerateRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{6}
+	return fileDescriptor_eacb9df0ce5738fa, []int{7}
 }
 func (m *ApplicationSetGenerateRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -448,7 +528,7 @@ func (m *ApplicationSetGenerateResponse) Reset()         { *m = ApplicationSetGe
 func (m *ApplicationSetGenerateResponse) String() string { return proto.CompactTextString(m) }
 func (*ApplicationSetGenerateResponse) ProtoMessage()    {}
 func (*ApplicationSetGenerateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_eacb9df0ce5738fa, []int{7}
+	return fileDescriptor_eacb9df0ce5738fa, []int{8}
 }
 func (m *ApplicationSetGenerateResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -487,6 +567,7 @@ func (m *ApplicationSetGenerateResponse) GetApplications() []*v1alpha1.Applicati
 func init() {
 	proto.RegisterType((*ApplicationSetGetQuery)(nil), "applicationset.ApplicationSetGetQuery")
 	proto.RegisterType((*ApplicationSetListQuery)(nil), "applicationset.ApplicationSetListQuery")
+	proto.RegisterType((*ApplicationSetWatchQuery)(nil), "applicationset.ApplicationSetWatchQuery")
 	proto.RegisterType((*ApplicationSetResponse)(nil), "applicationset.ApplicationSetResponse")
 	proto.RegisterType((*ApplicationSetCreateRequest)(nil), "applicationset.ApplicationSetCreateRequest")
 	proto.RegisterType((*ApplicationSetDeleteRequest)(nil), "applicationset.ApplicationSetDeleteRequest")
@@ -500,52 +581,58 @@ func init() {
 }
 
 var fileDescriptor_eacb9df0ce5738fa = []byte{
-	// 720 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x96, 0xcf, 0x6b, 0x13, 0x41,
-	0x14, 0xc7, 0x99, 0xb6, 0xc4, 0x74, 0x5a, 0x14, 0x06, 0x6c, 0x63, 0xb4, 0x31, 0x2c, 0xf4, 0x87,
-	0xad, 0x99, 0x25, 0xad, 0x07, 0xa9, 0x27, 0x7f, 0x51, 0x0a, 0x45, 0x74, 0x23, 0x0a, 0x7a, 0x90,
-	0xed, 0xe6, 0xb1, 0x5d, 0x9b, 0xec, 0x8c, 0x33, 0x93, 0x85, 0x52, 0xbc, 0x08, 0x1e, 0xf4, 0xe2,
-	0x41, 0xf4, 0x0f, 0xd0, 0x8b, 0x7f, 0x80, 0x77, 0x0f, 0x5e, 0x3c, 0x0a, 0xfe, 0x03, 0x52, 0xfc,
-	0x43, 0x64, 0x66, 0x37, 0x69, 0x77, 0x4c, 0x9a, 0x82, 0xd1, 0xdb, 0xbe, 0x9d, 0xd9, 0xf7, 0x3e,
-	0xf3, 0xde, 0x77, 0xbf, 0x0c, 0x5e, 0x96, 0x20, 0x12, 0x10, 0xae, 0xcf, 0x79, 0x2b, 0x0a, 0x7c,
-	0x15, 0xb1, 0x58, 0x82, 0xb2, 0x42, 0xca, 0x05, 0x53, 0x8c, 0x9c, 0xce, 0xbf, 0x2d, 0x5f, 0x08,
-	0x19, 0x0b, 0x5b, 0xe0, 0xfa, 0x3c, 0x72, 0xfd, 0x38, 0x66, 0x2a, 0x5d, 0x49, 0x77, 0x97, 0x9d,
-	0xdd, 0xab, 0x92, 0x46, 0xcc, 0xac, 0x06, 0x4c, 0x80, 0x9b, 0xd4, 0xdd, 0x10, 0x62, 0x10, 0xbe,
-	0x82, 0x66, 0xb6, 0x67, 0x2b, 0x8c, 0xd4, 0x4e, 0x67, 0x9b, 0x06, 0xac, 0xed, 0xfa, 0x22, 0x64,
-	0x5c, 0xb0, 0xa7, 0xe6, 0xa1, 0x16, 0x34, 0xdd, 0x64, 0xcd, 0xe5, 0xbb, 0xa1, 0xfe, 0x5e, 0x1e,
-	0xe5, 0x71, 0x93, 0xba, 0xdf, 0xe2, 0x3b, 0xfe, 0x1f, 0xd9, 0x9c, 0x07, 0x78, 0xe6, 0xfa, 0xe1,
-	0xbe, 0x06, 0xa8, 0x0d, 0x50, 0xf7, 0x3a, 0x20, 0xf6, 0x08, 0xc1, 0x13, 0xb1, 0xdf, 0x86, 0x12,
-	0xaa, 0xa2, 0xa5, 0x49, 0xcf, 0x3c, 0x93, 0x25, 0x7c, 0xc6, 0xe7, 0x5c, 0x82, 0xba, 0xe3, 0xb7,
-	0x41, 0x72, 0x3f, 0x80, 0xd2, 0x98, 0x59, 0xb6, 0x5f, 0x3b, 0xfb, 0x78, 0x36, 0x9f, 0x77, 0x2b,
-	0x92, 0x59, 0xe2, 0x32, 0x2e, 0x6a, 0x66, 0x08, 0x94, 0x2c, 0xa1, 0xea, 0xf8, 0xd2, 0xa4, 0xd7,
-	0x8b, 0xf5, 0x9a, 0x84, 0x16, 0x04, 0x8a, 0x89, 0x2c, 0x73, 0x2f, 0xee, 0x57, 0x7c, 0xbc, 0x7f,
-	0xf1, 0x4f, 0xc8, 0x3e, 0x95, 0x07, 0x92, 0xeb, 0x01, 0x90, 0x12, 0x3e, 0x95, 0x15, 0xcb, 0x0e,
-	0xd6, 0x0d, 0x89, 0xc2, 0xd6, 0xac, 0x0c, 0xc0, 0xd4, 0xea, 0x16, 0x3d, 0x6c, 0x38, 0xed, 0x36,
-	0xdc, 0x3c, 0x3c, 0x09, 0x9a, 0x34, 0x59, 0xa3, 0x7c, 0x37, 0xa4, 0xba, 0xe1, 0xf4, 0xc8, 0xe7,
-	0xb4, 0xdb, 0x70, 0x6a, 0x71, 0x58, 0x35, 0x9c, 0xaf, 0x08, 0x9f, 0xcf, 0x6f, 0xb9, 0x29, 0xc0,
-	0x57, 0xe0, 0xc1, 0xb3, 0x0e, 0xc8, 0x7e, 0x54, 0xe8, 0xdf, 0x53, 0x91, 0x19, 0x5c, 0xe8, 0x70,
-	0x09, 0x22, 0xed, 0x41, 0xd1, 0xcb, 0x22, 0xfd, 0xbe, 0x29, 0xf6, 0xbc, 0x4e, 0x6c, 0x3a, 0x5f,
-	0xf4, 0xb2, 0xc8, 0x79, 0x6c, 0x1f, 0xe2, 0x16, 0xb4, 0xe0, 0xf0, 0x10, 0x7f, 0x27, 0xa5, 0x87,
-	0xb6, 0x94, 0xee, 0x0b, 0x80, 0x51, 0x68, 0xf4, 0x1d, 0xc2, 0x73, 0xb6, 0xf8, 0xd3, 0xbf, 0xa3,
-	0x7f, 0xf7, 0x1b, 0xff, 0xa1, 0xfb, 0x0d, 0x50, 0xce, 0x1b, 0x84, 0x2b, 0x83, 0xb8, 0x32, 0x19,
-	0xb7, 0xf1, 0xf4, 0xd1, 0x91, 0x99, 0xff, 0x68, 0x6a, 0x75, 0x73, 0x64, 0x58, 0x5e, 0x2e, 0xfd,
-	0xea, 0xab, 0x49, 0x7c, 0x36, 0x4f, 0xd4, 0x00, 0x91, 0x44, 0x01, 0x90, 0x8f, 0x08, 0x8f, 0x6f,
-	0x80, 0x22, 0x0b, 0xd4, 0xb2, 0xbf, 0xfe, 0xae, 0x52, 0x1e, 0x69, 0xe7, 0x9c, 0x85, 0x17, 0x3f,
-	0x7e, 0xbd, 0x1d, 0xab, 0x92, 0x8a, 0x71, 0xcc, 0xa4, 0x6e, 0x79, 0xb0, 0x74, 0xf7, 0xb5, 0x24,
-	0x9e, 0x93, 0xf7, 0x08, 0x17, 0xbb, 0x3d, 0x24, 0xb5, 0x61, 0xa8, 0x39, 0x0d, 0x94, 0xe9, 0x49,
-	0xb7, 0xa7, 0xa3, 0x71, 0x56, 0x0c, 0xd3, 0xbc, 0x53, 0x1d, 0xc4, 0xd4, 0xb5, 0xe0, 0x75, 0xb4,
-	0x4c, 0x3e, 0x20, 0x3c, 0xa1, 0x9d, 0x91, 0x2c, 0x1e, 0x5f, 0xa5, 0xe7, 0x9e, 0xe5, 0xbb, 0xa3,
-	0x6c, 0xa0, 0x4e, 0xeb, 0x5c, 0x34, 0xc0, 0xe7, 0xc8, 0xec, 0x00, 0x60, 0xf2, 0x19, 0xe1, 0x42,
-	0xea, 0x4a, 0x64, 0xe5, 0x78, 0xcc, 0x9c, 0x77, 0x8d, 0x78, 0xd6, 0xae, 0xc1, 0xbc, 0xe4, 0x0c,
-	0xc2, 0x5c, 0xb7, 0x4d, 0xec, 0x25, 0xc2, 0x85, 0xd4, 0x87, 0x86, 0x61, 0xe7, 0xdc, 0xaa, 0x3c,
-	0x44, 0xca, 0xbd, 0x41, 0x67, 0xe2, 0x5b, 0x1e, 0x26, 0xbe, 0x2f, 0x08, 0x4f, 0x7b, 0x20, 0x59,
-	0x47, 0x04, 0xa0, 0xad, 0x6b, 0xd8, 0xac, 0x7b, 0xf6, 0x36, 0xda, 0x59, 0xeb, 0xb4, 0xce, 0x15,
-	0xc3, 0x4c, 0xc9, 0xe5, 0xe3, 0x99, 0x5d, 0x91, 0xf1, 0xd6, 0x94, 0x06, 0x7e, 0x8d, 0x30, 0xd1,
-	0x52, 0xe9, 0x9e, 0xe2, 0x76, 0x02, 0xb1, 0x92, 0x27, 0xfe, 0xe7, 0xe7, 0x68, 0x7a, 0xad, 0xd1,
-	0xa8, 0x54, 0x5f, 0x6b, 0x68, 0x52, 0xa7, 0x26, 0x87, 0xd1, 0x5f, 0xcd, 0x30, 0x2d, 0x92, 0xf9,
-	0x21, 0x4c, 0x60, 0xaa, 0xde, 0xd8, 0xfc, 0x76, 0x50, 0x41, 0xdf, 0x0f, 0x2a, 0xe8, 0xe7, 0x41,
-	0x05, 0x3d, 0xba, 0x76, 0xb2, 0xdb, 0x50, 0xd0, 0x8a, 0x20, 0xb6, 0xaf, 0x68, 0xdb, 0x05, 0x73,
-	0x07, 0x5a, 0xfb, 0x1d, 0x00, 0x00, 0xff, 0xff, 0xae, 0xa7, 0x68, 0xe3, 0xd1, 0x09, 0x00, 0x00,
+	// 801 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x96, 0x4b, 0x6b, 0x14, 0x4b,
+	0x14, 0xc7, 0xa9, 0x4c, 0x32, 0x77, 0x52, 0x09, 0xf7, 0x42, 0xc1, 0x4d, 0xc6, 0xd6, 0x8c, 0x43,
+	0x43, 0x92, 0x31, 0x71, 0xaa, 0x9d, 0xc4, 0x85, 0xc4, 0x95, 0x2f, 0x42, 0x20, 0x88, 0xf6, 0x48,
+	0x02, 0xba, 0x90, 0x4e, 0xcf, 0x61, 0xd2, 0x66, 0xa6, 0xbb, 0xad, 0xaa, 0x69, 0x08, 0xc1, 0x8d,
+	0xe0, 0xc6, 0x8d, 0x0b, 0xd1, 0x0f, 0xa0, 0x1b, 0xf7, 0xba, 0x72, 0x93, 0x85, 0x1b, 0x97, 0x82,
+	0x5f, 0x40, 0x82, 0x1f, 0x44, 0xaa, 0xba, 0xe7, 0xd1, 0x95, 0x79, 0x04, 0x1c, 0xdd, 0x75, 0x3d,
+	0xfa, 0xd4, 0xaf, 0xce, 0xf9, 0xd7, 0xbf, 0x0a, 0xaf, 0x70, 0x60, 0x11, 0x30, 0xcb, 0x09, 0xc3,
+	0x86, 0xe7, 0x3a, 0xc2, 0x0b, 0x7c, 0x0e, 0x42, 0x6b, 0xd2, 0x90, 0x05, 0x22, 0x20, 0xff, 0xa6,
+	0x7b, 0x8d, 0x0b, 0xf5, 0x20, 0xa8, 0x37, 0xc0, 0x72, 0x42, 0xcf, 0x72, 0x7c, 0x3f, 0x10, 0xf1,
+	0x48, 0x3c, 0xdb, 0x30, 0x0f, 0xae, 0x71, 0xea, 0x05, 0x6a, 0xd4, 0x0d, 0x18, 0x58, 0x51, 0xc5,
+	0xaa, 0x83, 0x0f, 0xcc, 0x11, 0x50, 0x4b, 0xe6, 0x6c, 0xd7, 0x3d, 0xb1, 0xdf, 0xda, 0xa3, 0x6e,
+	0xd0, 0xb4, 0x1c, 0x56, 0x0f, 0x42, 0x16, 0x3c, 0x51, 0x1f, 0x65, 0xb7, 0x66, 0x45, 0xeb, 0x56,
+	0x78, 0x50, 0x97, 0xff, 0xf3, 0x5e, 0x1e, 0x2b, 0xaa, 0x38, 0x8d, 0x70, 0xdf, 0x39, 0x15, 0xcd,
+	0xdc, 0xc1, 0x73, 0x37, 0xba, 0xf3, 0xaa, 0x20, 0x36, 0x41, 0xdc, 0x6f, 0x01, 0x3b, 0x24, 0x04,
+	0x4f, 0xfa, 0x4e, 0x13, 0xf2, 0xa8, 0x88, 0x4a, 0xd3, 0xb6, 0xfa, 0x26, 0x25, 0xfc, 0x9f, 0x13,
+	0x86, 0x1c, 0xc4, 0x5d, 0xa7, 0x09, 0x3c, 0x74, 0x5c, 0xc8, 0x4f, 0xa8, 0x61, 0xbd, 0xdb, 0x3c,
+	0xc2, 0xf3, 0xe9, 0xb8, 0xdb, 0x1e, 0x4f, 0x02, 0x1b, 0x38, 0x27, 0x99, 0xc1, 0x15, 0x3c, 0x8f,
+	0x8a, 0x99, 0xd2, 0xb4, 0xdd, 0x69, 0xcb, 0x31, 0x0e, 0x0d, 0x70, 0x45, 0xc0, 0x92, 0xc8, 0x9d,
+	0x76, 0xbf, 0xc5, 0x33, 0xfd, 0x17, 0xff, 0x8c, 0x70, 0x3e, 0xbd, 0xfa, 0xae, 0x23, 0xdc, 0xfd,
+	0xc1, 0xfb, 0xea, 0x45, 0x9a, 0x18, 0x82, 0x94, 0xe9, 0x8b, 0x54, 0xed, 0x45, 0x9a, 0xec, 0x20,
+	0xf5, 0x76, 0xcb, 0x99, 0x0c, 0x78, 0xd0, 0x62, 0x2e, 0xec, 0x00, 0xe3, 0x5e, 0xe0, 0xe7, 0xa7,
+	0xe2, 0x99, 0x5a, 0xb7, 0xf9, 0x01, 0xe9, 0x25, 0xb1, 0x81, 0x87, 0x52, 0x3d, 0x24, 0x8f, 0xff,
+	0x49, 0xb0, 0x12, 0xfa, 0x76, 0x93, 0x08, 0xac, 0x09, 0x4d, 0x65, 0x6f, 0x66, 0x6d, 0x9b, 0x76,
+	0xd5, 0x42, 0xdb, 0x6a, 0x51, 0x1f, 0x8f, 0xdd, 0x1a, 0x8d, 0xd6, 0x69, 0x78, 0x50, 0xa7, 0x52,
+	0x2d, 0xb4, 0xe7, 0x77, 0xda, 0x56, 0x0b, 0xd5, 0x38, 0xb4, 0x35, 0xcc, 0x2f, 0x08, 0x9f, 0x4f,
+	0x4f, 0xb9, 0xc5, 0xc0, 0x11, 0x60, 0xc3, 0xd3, 0x16, 0xf0, 0x7e, 0x54, 0xe8, 0xcf, 0x53, 0x91,
+	0x39, 0x9c, 0x6d, 0x85, 0x1c, 0x58, 0x9c, 0x83, 0x9c, 0x9d, 0xb4, 0x64, 0x7f, 0x8d, 0x1d, 0xda,
+	0x2d, 0x5f, 0x95, 0x31, 0x67, 0x27, 0x2d, 0xf3, 0x91, 0xbe, 0x89, 0xdb, 0xd0, 0x80, 0xee, 0x26,
+	0x7e, 0xef, 0x1c, 0xec, 0xea, 0xe7, 0xe0, 0x01, 0x03, 0x18, 0xc7, 0x01, 0x7b, 0x83, 0xf0, 0x82,
+	0x7e, 0x72, 0xe3, 0xa3, 0xdd, 0x3f, 0xfb, 0xd5, 0xbf, 0x90, 0xfd, 0x2a, 0x08, 0xf3, 0x15, 0xc2,
+	0x85, 0x41, 0x5c, 0x89, 0x8c, 0x9b, 0x78, 0xb6, 0xb7, 0x64, 0xca, 0x04, 0x66, 0xd6, 0xb6, 0xc6,
+	0x86, 0x65, 0xa7, 0xc2, 0xaf, 0x1d, 0x63, 0xfc, 0x7f, 0x9a, 0xa8, 0x0a, 0x2c, 0xf2, 0x5c, 0x20,
+	0xef, 0x11, 0xce, 0x6c, 0x82, 0x20, 0x4b, 0x54, 0xf3, 0xee, 0xfe, 0x96, 0x68, 0x8c, 0x35, 0x73,
+	0xe6, 0xd2, 0xf3, 0xef, 0x3f, 0x5f, 0x4f, 0x14, 0x49, 0x41, 0xd9, 0x7d, 0x54, 0xd1, 0x2e, 0x10,
+	0x6e, 0x1d, 0x49, 0x49, 0x3c, 0x23, 0x6f, 0x11, 0xce, 0xb5, 0x73, 0x48, 0xca, 0xa3, 0x50, 0x53,
+	0x1a, 0x30, 0xe8, 0x59, 0xa7, 0xc7, 0xa5, 0x31, 0x57, 0x15, 0xd3, 0xa2, 0x59, 0x1c, 0xc4, 0xd4,
+	0xbe, 0x3f, 0x36, 0xd0, 0x0a, 0x79, 0x87, 0xf0, 0xa4, 0xb4, 0x75, 0xb2, 0x3c, 0x7c, 0x95, 0x8e,
+	0xf5, 0x1b, 0xf7, 0xc6, 0x99, 0x40, 0x19, 0xd6, 0xbc, 0xa8, 0x80, 0xcf, 0x91, 0xf9, 0x01, 0xc0,
+	0xe4, 0x13, 0xc2, 0xd9, 0xd8, 0x95, 0xc8, 0xea, 0x70, 0xcc, 0x94, 0x77, 0x8d, 0xb9, 0xd6, 0x96,
+	0xc2, 0xbc, 0x64, 0x0e, 0xc2, 0xdc, 0xd0, 0x4d, 0xec, 0x05, 0xc2, 0xd9, 0xd8, 0x87, 0x46, 0x61,
+	0xa7, 0xdc, 0xca, 0x18, 0x21, 0xe5, 0x4e, 0xa1, 0x13, 0xf1, 0xad, 0x8c, 0x12, 0xdf, 0x31, 0xc2,
+	0xb3, 0x76, 0x72, 0x43, 0x49, 0xeb, 0x1a, 0x55, 0xeb, 0x8e, 0xbd, 0x8d, 0xb7, 0xd6, 0x32, 0xac,
+	0x79, 0x55, 0x31, 0x53, 0x72, 0x79, 0x38, 0xb3, 0xd5, 0xbe, 0x51, 0xcb, 0x42, 0x02, 0xbf, 0x44,
+	0x98, 0x48, 0xa9, 0xb4, 0x77, 0x71, 0x27, 0x02, 0x5f, 0xf0, 0x33, 0x9f, 0xf9, 0x05, 0x1a, 0xbf,
+	0xc9, 0x24, 0x2a, 0x95, 0x6f, 0x32, 0x1a, 0x55, 0xa8, 0x8a, 0xa1, 0xf4, 0x57, 0x56, 0x4c, 0xcb,
+	0x64, 0x71, 0x04, 0x13, 0xc4, 0xab, 0x7e, 0x44, 0x78, 0x4a, 0xbd, 0x45, 0x48, 0x69, 0xf8, 0xfa,
+	0xdd, 0x07, 0x8b, 0xb1, 0x33, 0xce, 0x44, 0xaa, 0xb8, 0x0a, 0xff, 0xb4, 0xff, 0x70, 0xc1, 0xc0,
+	0x69, 0xea, 0x3b, 0xb8, 0x82, 0x6e, 0x6e, 0x7d, 0x3d, 0x29, 0xa0, 0x6f, 0x27, 0x05, 0xf4, 0xe3,
+	0xa4, 0x80, 0x1e, 0x5e, 0x3f, 0xdb, 0x03, 0xd4, 0x6d, 0x78, 0xe0, 0xeb, 0xaf, 0xe2, 0xbd, 0xac,
+	0x7a, 0x76, 0xae, 0xff, 0x0a, 0x00, 0x00, 0xff, 0xff, 0x03, 0xd8, 0x6f, 0x1e, 0x44, 0x0b, 0x00,
+	0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -574,6 +661,7 @@ type ApplicationSetServiceClient interface {
 	ResourceTree(ctx context.Context, in *ApplicationSetTreeQuery, opts ...grpc.CallOption) (*v1alpha1.ApplicationSetTree, error)
 	// ListResourceEvents returns a list of event resources
 	ListResourceEvents(ctx context.Context, in *ApplicationSetGetQuery, opts ...grpc.CallOption) (*v1.EventList, error)
+	Watch(ctx context.Context, in *ApplicationSetWatchQuery, opts ...grpc.CallOption) (ApplicationSetService_WatchClient, error)
 }
 
 type applicationSetServiceClient struct {
@@ -647,6 +735,38 @@ func (c *applicationSetServiceClient) ListResourceEvents(ctx context.Context, in
 	return out, nil
 }
 
+func (c *applicationSetServiceClient) Watch(ctx context.Context, in *ApplicationSetWatchQuery, opts ...grpc.CallOption) (ApplicationSetService_WatchClient, error) {
+	stream, err := c.cc.NewStream(ctx, &_ApplicationSetService_serviceDesc.Streams[0], "/applicationset.ApplicationSetService/Watch", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &applicationSetServiceWatchClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type ApplicationSetService_WatchClient interface {
+	Recv() (*v1alpha1.ApplicationSetWatchEvent, error)
+	grpc.ClientStream
+}
+
+type applicationSetServiceWatchClient struct {
+	grpc.ClientStream
+}
+
+func (x *applicationSetServiceWatchClient) Recv() (*v1alpha1.ApplicationSetWatchEvent, error) {
+	m := new(v1alpha1.ApplicationSetWatchEvent)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // ApplicationSetServiceServer is the server API for ApplicationSetService service.
 type ApplicationSetServiceServer interface {
 	// Get returns an applicationset by name
@@ -663,6 +783,7 @@ type ApplicationSetServiceServer interface {
 	ResourceTree(context.Context, *ApplicationSetTreeQuery) (*v1alpha1.ApplicationSetTree, error)
 	// ListResourceEvents returns a list of event resources
 	ListResourceEvents(context.Context, *ApplicationSetGetQuery) (*v1.EventList, error)
+	Watch(*ApplicationSetWatchQuery, ApplicationSetService_WatchServer) error
 }
 
 // UnimplementedApplicationSetServiceServer can be embedded to have forward compatible implementations.
@@ -690,6 +811,9 @@ func (*UnimplementedApplicationSetServiceServer) ResourceTree(ctx context.Contex
 func (*UnimplementedApplicationSetServiceServer) ListResourceEvents(ctx context.Context, req *ApplicationSetGetQuery) (*v1.EventList, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method ListResourceEvents not implemented")
 }
+func (*UnimplementedApplicationSetServiceServer) Watch(req *ApplicationSetWatchQuery, srv ApplicationSetService_WatchServer) error {
+	return status.Errorf(codes.Unimplemented, "method Watch not implemented")
+}
 
 func RegisterApplicationSetServiceServer(s *grpc.Server, srv ApplicationSetServiceServer) {
 	s.RegisterService(&_ApplicationSetService_serviceDesc, srv)
@@ -821,6 +945,27 @@ func _ApplicationSetService_ListResourceEvents_Handler(srv interface{}, ctx cont
 	return interceptor(ctx, in, info, handler)
 }
 
+func _ApplicationSetService_Watch_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(ApplicationSetWatchQuery)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(ApplicationSetServiceServer).Watch(m, &applicationSetServiceWatchServer{stream})
+}
+
+type ApplicationSetService_WatchServer interface {
+	Send(*v1alpha1.ApplicationSetWatchEvent) error
+	grpc.ServerStream
+}
+
+type applicationSetServiceWatchServer struct {
+	grpc.ServerStream
+}
+
+func (x *applicationSetServiceWatchServer) Send(m *v1alpha1.ApplicationSetWatchEvent) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 var _ApplicationSetService_serviceDesc = grpc.ServiceDesc{
 	ServiceName: "applicationset.ApplicationSetService",
 	HandlerType: (*ApplicationSetServiceServer)(nil),
@@ -854,7 +999,13 @@ var _ApplicationSetService_serviceDesc = grpc.ServiceDesc{
 			Handler:    _ApplicationSetService_ListResourceEvents_Handler,
 		},
 	},
-	Streams:  []grpc.StreamDesc{},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "Watch",
+			Handler:       _ApplicationSetService_Watch_Handler,
+			ServerStreams: true,
+		},
+	},
 	Metadata: "server/applicationset/applicationset.proto",
 }
 
@@ -949,6 +1100,70 @@ func (m *ApplicationSetListQuery) MarshalToSizedBuffer(dAtA []byte) (int, error)
 	return len(dAtA) - i, nil
 }
 
+func (m *ApplicationSetWatchQuery) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ApplicationSetWatchQuery) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ApplicationSetWatchQuery) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.ResourceVersion) > 0 {
+		i -= len(m.ResourceVersion)
+		copy(dAtA[i:], m.ResourceVersion)
+		i = encodeVarintApplicationset(dAtA, i, uint64(len(m.ResourceVersion)))
+		i--
+		dAtA[i] = 0x2a
+	}
+	if len(m.AppSetNamespace) > 0 {
+		i -= len(m.AppSetNamespace)
+		copy(dAtA[i:], m.AppSetNamespace)
+		i = encodeVarintApplicationset(dAtA, i, uint64(len(m.AppSetNamespace)))
+		i--
+		dAtA[i] = 0x22
+	}
+	if len(m.Selector) > 0 {
+		i -= len(m.Selector)
+		copy(dAtA[i:], m.Selector)
+		i = encodeVarintApplicationset(dAtA, i, uint64(len(m.Selector)))
+		i--
+		dAtA[i] = 0x1a
+	}
+	if len(m.Projects) > 0 {
+		for iNdEx := len(m.Projects) - 1; iNdEx >= 0; iNdEx-- {
+			i -= len(m.Projects[iNdEx])
+			copy(dAtA[i:], m.Projects[iNdEx])
+			i = encodeVarintApplicationset(dAtA, i, uint64(len(m.Projects[iNdEx])))
+			i--
+			dAtA[i] = 0x12
+		}
+	}
+	if len(m.Name) > 0 {
+		i -= len(m.Name)
+		copy(dAtA[i:], m.Name)
+		i = encodeVarintApplicationset(dAtA, i, uint64(len(m.Name)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *ApplicationSetResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1273,6 +1488,40 @@ func (m *ApplicationSetListQuery) Size() (n int) {
 	return n
 }
 
+func (m *ApplicationSetWatchQuery) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Name)
+	if l > 0 {
+		n += 1 + l + sovApplicationset(uint64(l))
+	}
+	if len(m.Projects) > 0 {
+		for _, s := range m.Projects {
+			l = len(s)
+			n += 1 + l + sovApplicationset(uint64(l))
+		}
+	}
+	l = len(m.Selector)
+	if l > 0 {
+		n += 1 + l + sovApplicationset(uint64(l))
+	}
+	l = len(m.AppSetNamespace)
+	if l > 0 {
+		n += 1 + l + sovApplicationset(uint64(l))
+	}
+	l = len(m.ResourceVersion)
+	if l > 0 {
+		n += 1 + l + sovApplicationset(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *ApplicationSetResponse) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1657,6 +1906,217 @@ func (m *ApplicationSetListQuery) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *ApplicationSetWatchQuery) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowApplicationset
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ApplicationSetWatchQuery: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ApplicationSetWatchQuery: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApplicationset
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Name = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Projects", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApplicationset
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Projects = append(m.Projects, string(dAtA[iNdEx:postIndex]))
+			iNdEx = postIndex
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Selector", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApplicationset
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Selector = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field AppSetNamespace", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApplicationset
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.AppSetNamespace = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 5:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ResourceVersion", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowApplicationset
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ResourceVersion = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipApplicationset(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthApplicationset
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *ApplicationSetResponse) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0

pkg/apiclient/applicationset/applicationset.pb.gw.go

@@ -443,6 +443,34 @@ func local_request_ApplicationSetService_ListResourceEvents_0(ctx context.Contex
 
 }
 
+var (
+	filter_ApplicationSetService_Watch_0 = &utilities.DoubleArray{Encoding: map[string]int{}, Base: []int(nil), Check: []int(nil)}
+)
+
+func request_ApplicationSetService_Watch_0(ctx context.Context, marshaler runtime.Marshaler, client ApplicationSetServiceClient, req *http.Request, pathParams map[string]string) (ApplicationSetService_WatchClient, runtime.ServerMetadata, error) {
+	var protoReq ApplicationSetWatchQuery
+	var metadata runtime.ServerMetadata
+
+	if err := req.ParseForm(); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if err := runtime.PopulateQueryParameters(&protoReq, req.Form, filter_ApplicationSetService_Watch_0); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+
+	stream, err := client.Watch(ctx, &protoReq)
+	if err != nil {
+		return nil, metadata, err
+	}
+	header, err := stream.Header()
+	if err != nil {
+		return nil, metadata, err
+	}
+	metadata.HeaderMD = header
+	return stream, metadata, nil
+
+}
+
 // RegisterApplicationSetServiceHandlerServer registers the http handlers for service ApplicationSetService to "mux".
 // UnaryRPC     :call ApplicationSetServiceServer directly.
 // StreamingRPC :currently unsupported pending https://github.com/grpc/grpc-go/issues/906.
@@ -610,6 +638,13 @@ func RegisterApplicationSetServiceHandlerServer(ctx context.Context, mux *runtim
 
 	})
 
+	mux.Handle("GET", pattern_ApplicationSetService_Watch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		err := status.Error(codes.Unimplemented, "streaming calls are not yet supported in the in-process transport")
+		_, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+		return
+	})
+
 	return nil
 }
 
@@ -791,6 +826,26 @@ func RegisterApplicationSetServiceHandlerClient(ctx context.Context, mux *runtim
 
 	})
 
+	mux.Handle("GET", pattern_ApplicationSetService_Watch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		rctx, err := runtime.AnnotateContext(ctx, mux, req)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_ApplicationSetService_Watch_0(rctx, inboundMarshaler, client, req, pathParams)
+		ctx = runtime.NewServerMetadataContext(ctx, md)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_ApplicationSetService_Watch_0(ctx, mux, outboundMarshaler, w, req, func() (proto.Message, error) { return resp.Recv() }, mux.GetForwardResponseOptions()...)
+
+	})
+
 	return nil
 }
 
@@ -808,6 +863,8 @@ var (
 	pattern_ApplicationSetService_ResourceTree_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 3, 2, 4}, []string{"api", "v1", "applicationsets", "name", "resource-tree"}, "", runtime.AssumeColonVerbOpt(true)))
 
 	pattern_ApplicationSetService_ListResourceEvents_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 3, 2, 4}, []string{"api", "v1", "applicationsets", "name", "events"}, "", runtime.AssumeColonVerbOpt(true)))
+
+	pattern_ApplicationSetService_Watch_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 2, 3}, []string{"api", "v1", "stream", "applicationsets"}, "", runtime.AssumeColonVerbOpt(true)))
 )
 
 var (
@@ -824,4 +881,6 @@ var (
 	forward_ApplicationSetService_ResourceTree_0 = runtime.ForwardResponseMessage
 
 	forward_ApplicationSetService_ListResourceEvents_0 = runtime.ForwardResponseMessage
+
+	forward_ApplicationSetService_Watch_0 = runtime.ForwardResponseStream
 )

pkg/apiclient/applicationset/forwarder_overwrite.go

@@ -0,0 +1,22 @@
+package applicationset
+
+import (
+	"errors"
+
+	"github.com/argoproj/pkg/v2/grpc/http"
+
+	//nolint:staticcheck
+	"github.com/golang/protobuf/proto"
+
+	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
+)
+
+func init() {
+	forward_ApplicationSetService_Watch_0 = http.NewStreamForwarder(func(message proto.Message) (string, error) {
+		event, ok := message.(*v1alpha1.ApplicationSetWatchEvent)
+		if !ok {
+			return "", errors.New("unexpected message type")
+		}
+		return event.ApplicationSet.Name, nil
+	})
+}

pkg/apis/application/v1alpha1/applicationset_types.go

@@ -24,6 +24,7 @@ import (
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/watch"
 
 	"github.com/argoproj/argo-cd/v3/common"
 	"github.com/argoproj/argo-cd/v3/util/security"
@@ -1077,3 +1078,17 @@ func (a *ApplicationSet) QualifiedName() string {
 	}
 	return a.Namespace + "/" + a.Name
 }
+
+// ApplicationSetWatchEvent contains information about application change.
+type ApplicationSetWatchEvent struct {
+	// Type represents the Kubernetes watch event type. The protobuf tag uses
+	// casttype to ensure the generated Go code keeps this field as
+	// watch.EventType (a strong Go type) instead of falling back to a plain string
+	Type watch.EventType `json:"type" protobuf:"bytes,1,opt,name=type,casttype=k8s.io/apimachinery/pkg/watch.EventType"`
+	// ApplicationSet is:
+	//  * If Type is Added or Modified: the new state of the object.
+	//  * If Type is Deleted: the state of the object immediately before deletion.
+	//  * If Type is Error: *api.Status is recommended; other types may make sense
+	//    depending on context
+	ApplicationSet ApplicationSet `json:"applicationSet" protobuf:"bytes,2,opt,name=applicationSet"`
+}

pkg/apis/application/v1alpha1/generated.proto

@@ -459,6 +459,21 @@ message ApplicationSetTree {
   repeated ResourceNode nodes = 1;
 }
 
+// ApplicationSetWatchEvent contains information about application change.
+message ApplicationSetWatchEvent {
+  // Type represents the Kubernetes watch event type. The protobuf tag uses
+  // casttype to ensure the generated Go code keeps this field as
+  // watch.EventType (a strong Go type) instead of falling back to a plain string
+  optional string type = 1;
+
+  // ApplicationSet is:
+  //  * If Type is Added or Modified: the new state of the object.
+  //  * If Type is Deleted: the state of the object immediately before deletion.
+  //  * If Type is Error: *api.Status is recommended; other types may make sense
+  //    depending on context
+  optional ApplicationSet applicationSet = 2;
+}
+
 // ApplicationSource contains all required information about the source of an application
 message ApplicationSource {
   // RepoURL is the URL to the repository (Git or Helm) that contains the application manifests

pkg/apis/application/v1alpha1/zz_generated.deepcopy.go

@@ -1044,6 +1044,23 @@ func (in *ApplicationSetTree) DeepCopy() *ApplicationSetTree {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ApplicationSetWatchEvent) DeepCopyInto(out *ApplicationSetWatchEvent) {
+	*out = *in
+	in.ApplicationSet.DeepCopyInto(&out.ApplicationSet)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ApplicationSetWatchEvent.
+func (in *ApplicationSetWatchEvent) DeepCopy() *ApplicationSetWatchEvent {
+	if in == nil {
+		return nil
+	}
+	out := new(ApplicationSetWatchEvent)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ApplicationSource) DeepCopyInto(out *ApplicationSource) {
 	*out = *in

server/application/application.go

@@ -47,6 +47,7 @@ import (
 	appclientset "github.com/argoproj/argo-cd/v3/pkg/client/clientset/versioned"
 	applisters "github.com/argoproj/argo-cd/v3/pkg/client/listers/application/v1alpha1"
 	"github.com/argoproj/argo-cd/v3/reposerver/apiclient"
+	"github.com/argoproj/argo-cd/v3/server/broadcast"
 	servercache "github.com/argoproj/argo-cd/v3/server/cache"
 	"github.com/argoproj/argo-cd/v3/server/deeplinks"
 	applog "github.com/argoproj/argo-cd/v3/util/app/log"
@@ -90,7 +91,7 @@ type Server struct {
 	appclientset           appclientset.Interface
 	appLister              applisters.ApplicationLister
 	appInformer            cache.SharedIndexInformer
-	appBroadcaster         Broadcaster
+	appBroadcaster         broadcast.Broadcaster[v1alpha1.ApplicationWatchEvent]
 	repoClientset          apiclient.Clientset
 	kubectl                kube.Kubectl
 	db                     db.ArgoDB
@@ -111,7 +112,7 @@ func NewServer(
 	appclientset appclientset.Interface,
 	appLister applisters.ApplicationLister,
 	appInformer cache.SharedIndexInformer,
-	appBroadcaster Broadcaster,
+	appBroadcaster broadcast.Broadcaster[v1alpha1.ApplicationWatchEvent],
 	repoClientset apiclient.Clientset,
 	cache *servercache.Cache,
 	kubectl kube.Kubectl,
@@ -125,8 +126,15 @@ func NewServer(
 	syncWithReplaceAllowed bool,
 ) (application.ApplicationServiceServer, AppResourceTreeFn) {
 	if appBroadcaster == nil {
-		appBroadcaster = &broadcasterHandler{}
+		appBroadcaster = broadcast.NewHandler[v1alpha1.Application, v1alpha1.ApplicationWatchEvent](
+			func(app *v1alpha1.Application, eventType watch.EventType) *v1alpha1.ApplicationWatchEvent {
+				return &v1alpha1.ApplicationWatchEvent{Application: *app, Type: eventType}
+			},
+			applog.GetAppLogFields,
+		)
 	}
+	// Register Application-level broadcaster to receive create/update/delete events
+	// and handle general application event processing.
 	_, err := appInformer.AddEventHandler(appBroadcaster)
 	if err != nil {
 		log.Error(err)

server/application/broadcaster.go

@@ -1,96 +0,0 @@
-package application
-
-import (
-	"sync"
-
-	log "github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/watch"
-
-	appv1 "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
-	applog "github.com/argoproj/argo-cd/v3/util/app/log"
-)
-
-type subscriber struct {
-	ch      chan *appv1.ApplicationWatchEvent
-	filters []func(*appv1.ApplicationWatchEvent) bool
-}
-
-func (s *subscriber) matches(event *appv1.ApplicationWatchEvent) bool {
-	for i := range s.filters {
-		if !s.filters[i](event) {
-			return false
-		}
-	}
-	return true
-}
-
-// Broadcaster is an interface for broadcasting application informer watch events to multiple subscribers.
-type Broadcaster interface {
-	Subscribe(ch chan *appv1.ApplicationWatchEvent, filters ...func(event *appv1.ApplicationWatchEvent) bool) func()
-	OnAdd(any, bool)
-	OnUpdate(any, any)
-	OnDelete(any)
-}
-
-type broadcasterHandler struct {
-	lock        sync.Mutex
-	subscribers []*subscriber
-}
-
-func (b *broadcasterHandler) notify(event *appv1.ApplicationWatchEvent) {
-	// Make a local copy of b.subscribers, then send channel events outside the lock,
-	// to avoid data race on b.subscribers changes
-	subscribers := []*subscriber{}
-	b.lock.Lock()
-	subscribers = append(subscribers, b.subscribers...)
-	b.lock.Unlock()
-
-	for _, s := range subscribers {
-		if s.matches(event) {
-			select {
-			case s.ch <- event:
-			default:
-				// drop event if cannot send right away
-				log.WithFields(applog.GetAppLogFields(&event.Application)).Warn("unable to send event notification")
-			}
-		}
-	}
-}
-
-// Subscribe forward application informer watch events to the provided channel.
-// The watch events are dropped if no receives are reading events from the channel so the channel must have
-// buffer if dropping events is not acceptable.
-func (b *broadcasterHandler) Subscribe(ch chan *appv1.ApplicationWatchEvent, filters ...func(event *appv1.ApplicationWatchEvent) bool) func() {
-	b.lock.Lock()
-	defer b.lock.Unlock()
-	subscriber := &subscriber{ch, filters}
-	b.subscribers = append(b.subscribers, subscriber)
-	return func() {
-		b.lock.Lock()
-		defer b.lock.Unlock()
-		for i := range b.subscribers {
-			if b.subscribers[i] == subscriber {
-				b.subscribers = append(b.subscribers[:i], b.subscribers[i+1:]...)
-				break
-			}
-		}
-	}
-}
-
-func (b *broadcasterHandler) OnAdd(obj any, _ bool) {
-	if app, ok := obj.(*appv1.Application); ok {
-		b.notify(&appv1.ApplicationWatchEvent{Application: *app, Type: watch.Added})
-	}
-}
-
-func (b *broadcasterHandler) OnUpdate(_, newObj any) {
-	if app, ok := newObj.(*appv1.Application); ok {
-		b.notify(&appv1.ApplicationWatchEvent{Application: *app, Type: watch.Modified})
-	}
-}
-
-func (b *broadcasterHandler) OnDelete(obj any) {
-	if app, ok := obj.(*appv1.Application); ok {
-		b.notify(&appv1.ApplicationWatchEvent{Application: *app, Type: watch.Deleted})
-	}
-}

server/application/mocks/Broadcaster.go

@@ -1,241 +0,0 @@
-// Code generated by mockery; DO NOT EDIT.
-// github.com/vektra/mockery
-// template: testify
-
-package mocks
-
-import (
-	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
-	mock "github.com/stretchr/testify/mock"
-)
-
-// NewBroadcaster creates a new instance of Broadcaster. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-// The first argument is typically a *testing.T value.
-func NewBroadcaster(t interface {
-	mock.TestingT
-	Cleanup(func())
-}) *Broadcaster {
-	mock := &Broadcaster{}
-	mock.Mock.Test(t)
-
-	t.Cleanup(func() { mock.AssertExpectations(t) })
-
-	return mock
-}
-
-// Broadcaster is an autogenerated mock type for the Broadcaster type
-type Broadcaster struct {
-	mock.Mock
-}
-
-type Broadcaster_Expecter struct {
-	mock *mock.Mock
-}
-
-func (_m *Broadcaster) EXPECT() *Broadcaster_Expecter {
-	return &Broadcaster_Expecter{mock: &_m.Mock}
-}
-
-// OnAdd provides a mock function for the type Broadcaster
-func (_mock *Broadcaster) OnAdd(v any, b bool) {
-	_mock.Called(v, b)
-	return
-}
-
-// Broadcaster_OnAdd_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnAdd'
-type Broadcaster_OnAdd_Call struct {
-	*mock.Call
-}
-
-// OnAdd is a helper method to define mock.On call
-//   - v any
-//   - b bool
-func (_e *Broadcaster_Expecter) OnAdd(v interface{}, b interface{}) *Broadcaster_OnAdd_Call {
-	return &Broadcaster_OnAdd_Call{Call: _e.mock.On("OnAdd", v, b)}
-}
-
-func (_c *Broadcaster_OnAdd_Call) Run(run func(v any, b bool)) *Broadcaster_OnAdd_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		var arg0 any
-		if args[0] != nil {
-			arg0 = args[0].(any)
-		}
-		var arg1 bool
-		if args[1] != nil {
-			arg1 = args[1].(bool)
-		}
-		run(
-			arg0,
-			arg1,
-		)
-	})
-	return _c
-}
-
-func (_c *Broadcaster_OnAdd_Call) Return() *Broadcaster_OnAdd_Call {
-	_c.Call.Return()
-	return _c
-}
-
-func (_c *Broadcaster_OnAdd_Call) RunAndReturn(run func(v any, b bool)) *Broadcaster_OnAdd_Call {
-	_c.Run(run)
-	return _c
-}
-
-// OnDelete provides a mock function for the type Broadcaster
-func (_mock *Broadcaster) OnDelete(v any) {
-	_mock.Called(v)
-	return
-}
-
-// Broadcaster_OnDelete_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnDelete'
-type Broadcaster_OnDelete_Call struct {
-	*mock.Call
-}
-
-// OnDelete is a helper method to define mock.On call
-//   - v any
-func (_e *Broadcaster_Expecter) OnDelete(v interface{}) *Broadcaster_OnDelete_Call {
-	return &Broadcaster_OnDelete_Call{Call: _e.mock.On("OnDelete", v)}
-}
-
-func (_c *Broadcaster_OnDelete_Call) Run(run func(v any)) *Broadcaster_OnDelete_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		var arg0 any
-		if args[0] != nil {
-			arg0 = args[0].(any)
-		}
-		run(
-			arg0,
-		)
-	})
-	return _c
-}
-
-func (_c *Broadcaster_OnDelete_Call) Return() *Broadcaster_OnDelete_Call {
-	_c.Call.Return()
-	return _c
-}
-
-func (_c *Broadcaster_OnDelete_Call) RunAndReturn(run func(v any)) *Broadcaster_OnDelete_Call {
-	_c.Run(run)
-	return _c
-}
-
-// OnUpdate provides a mock function for the type Broadcaster
-func (_mock *Broadcaster) OnUpdate(v any, v1 any) {
-	_mock.Called(v, v1)
-	return
-}
-
-// Broadcaster_OnUpdate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnUpdate'
-type Broadcaster_OnUpdate_Call struct {
-	*mock.Call
-}
-
-// OnUpdate is a helper method to define mock.On call
-//   - v any
-//   - v1 any
-func (_e *Broadcaster_Expecter) OnUpdate(v interface{}, v1 interface{}) *Broadcaster_OnUpdate_Call {
-	return &Broadcaster_OnUpdate_Call{Call: _e.mock.On("OnUpdate", v, v1)}
-}
-
-func (_c *Broadcaster_OnUpdate_Call) Run(run func(v any, v1 any)) *Broadcaster_OnUpdate_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		var arg0 any
-		if args[0] != nil {
-			arg0 = args[0].(any)
-		}
-		var arg1 any
-		if args[1] != nil {
-			arg1 = args[1].(any)
-		}
-		run(
-			arg0,
-			arg1,
-		)
-	})
-	return _c
-}
-
-func (_c *Broadcaster_OnUpdate_Call) Return() *Broadcaster_OnUpdate_Call {
-	_c.Call.Return()
-	return _c
-}
-
-func (_c *Broadcaster_OnUpdate_Call) RunAndReturn(run func(v any, v1 any)) *Broadcaster_OnUpdate_Call {
-	_c.Run(run)
-	return _c
-}
-
-// Subscribe provides a mock function for the type Broadcaster
-func (_mock *Broadcaster) Subscribe(ch chan *v1alpha1.ApplicationWatchEvent, filters ...func(event *v1alpha1.ApplicationWatchEvent) bool) func() {
-	// func(event *v1alpha1.ApplicationWatchEvent) bool
-	_va := make([]interface{}, len(filters))
-	for _i := range filters {
-		_va[_i] = filters[_i]
-	}
-	var _ca []interface{}
-	_ca = append(_ca, ch)
-	_ca = append(_ca, _va...)
-	ret := _mock.Called(_ca...)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Subscribe")
-	}
-
-	var r0 func()
-	if returnFunc, ok := ret.Get(0).(func(chan *v1alpha1.ApplicationWatchEvent, ...func(event *v1alpha1.ApplicationWatchEvent) bool) func()); ok {
-		r0 = returnFunc(ch, filters...)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(func())
-		}
-	}
-	return r0
-}
-
-// Broadcaster_Subscribe_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Subscribe'
-type Broadcaster_Subscribe_Call struct {
-	*mock.Call
-}
-
-// Subscribe is a helper method to define mock.On call
-//   - ch chan *v1alpha1.ApplicationWatchEvent
-//   - filters ...func(event *v1alpha1.ApplicationWatchEvent) bool
-func (_e *Broadcaster_Expecter) Subscribe(ch interface{}, filters ...interface{}) *Broadcaster_Subscribe_Call {
-	return &Broadcaster_Subscribe_Call{Call: _e.mock.On("Subscribe",
-		append([]interface{}{ch}, filters...)...)}
-}
-
-func (_c *Broadcaster_Subscribe_Call) Run(run func(ch chan *v1alpha1.ApplicationWatchEvent, filters ...func(event *v1alpha1.ApplicationWatchEvent) bool)) *Broadcaster_Subscribe_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		var arg0 chan *v1alpha1.ApplicationWatchEvent
-		if args[0] != nil {
-			arg0 = args[0].(chan *v1alpha1.ApplicationWatchEvent)
-		}
-		var arg1 []func(event *v1alpha1.ApplicationWatchEvent) bool
-		variadicArgs := make([]func(event *v1alpha1.ApplicationWatchEvent) bool, len(args)-1)
-		for i, a := range args[1:] {
-			if a != nil {
-				variadicArgs[i] = a.(func(event *v1alpha1.ApplicationWatchEvent) bool)
-			}
-		}
-		arg1 = variadicArgs
-		run(
-			arg0,
-			arg1...,
-		)
-	})
-	return _c
-}
-
-func (_c *Broadcaster_Subscribe_Call) Return(fn func()) *Broadcaster_Subscribe_Call {
-	_c.Call.Return(fn)
-	return _c
-}
-
-func (_c *Broadcaster_Subscribe_Call) RunAndReturn(run func(ch chan *v1alpha1.ApplicationWatchEvent, filters ...func(event *v1alpha1.ApplicationWatchEvent) bool) func()) *Broadcaster_Subscribe_Call {
-	_c.Call.Return(run)
-	return _c
-}

server/applicationset/applicationset.go

@@ -20,6 +20,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
@@ -30,11 +31,14 @@ import (
 	"github.com/argoproj/argo-cd/v3/applicationset/services"
 	appsetstatus "github.com/argoproj/argo-cd/v3/applicationset/status"
 	appsetutils "github.com/argoproj/argo-cd/v3/applicationset/utils"
+	argocommon "github.com/argoproj/argo-cd/v3/common"
 	"github.com/argoproj/argo-cd/v3/pkg/apiclient/applicationset"
 	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
 	appclientset "github.com/argoproj/argo-cd/v3/pkg/client/clientset/versioned"
 	applisters "github.com/argoproj/argo-cd/v3/pkg/client/listers/application/v1alpha1"
 	repoapiclient "github.com/argoproj/argo-cd/v3/reposerver/apiclient"
+	"github.com/argoproj/argo-cd/v3/server/broadcast"
+	applog "github.com/argoproj/argo-cd/v3/util/app/log"
 	"github.com/argoproj/argo-cd/v3/util/argo"
 	"github.com/argoproj/argo-cd/v3/util/collections"
 	"github.com/argoproj/argo-cd/v3/util/db"
@@ -56,6 +60,7 @@ type Server struct {
 	appclientset             appclientset.Interface
 	appsetInformer           cache.SharedIndexInformer
 	appsetLister             applisters.ApplicationSetLister
+	appSetBroadcaster        broadcast.Broadcaster[v1alpha1.ApplicationSetWatchEvent]
 	auditLogger              *argo.AuditLogger
 	projectLock              sync.KeyLock
 	enabledNamespaces        []string
@@ -68,6 +73,97 @@ type Server struct {
 	EnableGitHubAPIMetrics   bool
 }
 
+func (s *Server) Watch(q *applicationset.ApplicationSetWatchQuery, ws applicationset.ApplicationSetService_WatchServer) error {
+	appsetName := q.GetName()
+	appsetNs := q.GetAppSetNamespace()
+	logCtx := log.NewEntry(log.New())
+	if q.Name != "" {
+		logCtx = logCtx.WithField("applicationset", q.Name)
+	}
+	projects := map[string]bool{}
+	for _, project := range q.Projects {
+		projects[project] = true
+	}
+	claims := ws.Context().Value("claims")
+	selector, err := labels.Parse(q.GetSelector())
+	if err != nil {
+		return fmt.Errorf("error parsing labels with selectors: %w", err)
+	}
+	minVersion := 0
+	if q.GetResourceVersion() != "" {
+		if minVersion, err = strconv.Atoi(q.GetResourceVersion()); err != nil {
+			minVersion = 0
+		}
+	}
+	sendIfPermitted := func(a v1alpha1.ApplicationSet, eventType watch.EventType) {
+		permitted := s.isApplicationsetPermitted(selector, minVersion, claims, appsetName, appsetNs, projects, a)
+		if !permitted {
+			return
+		}
+		err := ws.Send(&v1alpha1.ApplicationSetWatchEvent{
+			Type:           eventType,
+			ApplicationSet: a,
+		})
+		if err != nil {
+			logCtx.Warnf("Unable to send stream message: %v", err)
+			return
+		}
+	}
+	events := make(chan *v1alpha1.ApplicationSetWatchEvent, argocommon.WatchAPIBufferSize)
+	// Subscribe before listing so that events arriving between list and subscribe are not lost
+	unsubscribe := s.appSetBroadcaster.Subscribe(events)
+	defer unsubscribe()
+	if q.GetName() != "" {
+		appsets, err := s.appsetLister.List(selector)
+		if err != nil {
+			return fmt.Errorf("error listing appsets with selector: %w", err)
+		}
+		sort.Slice(appsets, func(i, j int) bool {
+			return appsets[i].QualifiedName() < appsets[j].QualifiedName()
+		})
+		for i := range appsets {
+			sendIfPermitted(*appsets[i], watch.Added)
+		}
+	}
+	for {
+		select {
+		case event := <-events:
+			sendIfPermitted(event.ApplicationSet, event.Type)
+		case <-ws.Context().Done():
+			return nil
+		}
+	}
+}
+
+// isApplicationsetPermitted checks if an appset is permitted
+func (s *Server) isApplicationsetPermitted(selector labels.Selector, minVersion int, claims any, appsetName, appsetNs string, projects map[string]bool, appset v1alpha1.ApplicationSet) bool {
+	if len(projects) > 0 && !projects[appset.Spec.Template.Spec.Project] {
+		return false
+	}
+
+	if appsetVersion, err := strconv.Atoi(appset.ResourceVersion); err == nil && appsetVersion < minVersion {
+		return false
+	}
+	// Match by name, and optionally by namespace if provided
+	nameMatches := appsetName == "" || appset.Name == appsetName
+	nsMatches := appsetNs == "" || appset.Namespace == appsetNs
+	matchedEvent := nameMatches && nsMatches && selector.Matches(labels.Set(appset.Labels))
+	if !matchedEvent {
+		return false
+	}
+	// Skip any applicationsets that is neither in the control plane's namespace
+	// nor in the list of enabled namespaces.
+	if !security.IsNamespaceEnabled(appset.Namespace, s.ns, s.enabledNamespaces) {
+		return false
+	}
+
+	if !s.enf.Enforce(claims, rbac.ResourceApplicationSets, rbac.ActionGet, appset.RBACName(s.ns)) {
+		return false
+	}
+
+	return true
+}
+
 // NewServer returns a new instance of the ApplicationSet service
 func NewServer(
 	db db.ArgoDB,
@@ -79,6 +175,7 @@ func NewServer(
 	appclientset appclientset.Interface,
 	appsetInformer cache.SharedIndexInformer,
 	appsetLister applisters.ApplicationSetLister,
+	appSetBroadcaster broadcast.Broadcaster[v1alpha1.ApplicationSetWatchEvent],
 	namespace string,
 	projectLock sync.KeyLock,
 	enabledNamespaces []string,
@@ -91,6 +188,20 @@ func NewServer(
 	enableK8sEvent []string,
 	clusterInformer *settings.ClusterInformer,
 ) applicationset.ApplicationSetServiceServer {
+	if appSetBroadcaster == nil {
+		appSetBroadcaster = broadcast.NewHandler[v1alpha1.ApplicationSet, v1alpha1.ApplicationSetWatchEvent](
+			func(appset *v1alpha1.ApplicationSet, eventType watch.EventType) *v1alpha1.ApplicationSetWatchEvent {
+				return &v1alpha1.ApplicationSetWatchEvent{ApplicationSet: *appset, Type: eventType}
+			},
+			applog.GetAppSetLogFields,
+		)
+	}
+	// Register ApplicationSet level broadcaster to receive create/update/delete events
+	// and handle general applicationset event processing.
+	_, err := appsetInformer.AddEventHandler(appSetBroadcaster)
+	if err != nil {
+		log.Error(err)
+	}
 	s := &Server{
 		ns:                       namespace,
 		db:                       db,
@@ -102,6 +213,7 @@ func NewServer(
 		appclientset:             appclientset,
 		appsetInformer:           appsetInformer,
 		appsetLister:             appsetLister,
+		appSetBroadcaster:        appSetBroadcaster,
 		projectLock:              projectLock,
 		auditLogger:              argo.NewAuditLogger(kubeclientset, "argocd-server", enableK8sEvent),
 		enabledNamespaces:        enabledNamespaces,
@@ -141,7 +253,7 @@ func (s *Server) List(ctx context.Context, q *applicationset.ApplicationSetListQ
 
 	newItems := make([]v1alpha1.ApplicationSet, 0)
 	for _, a := range appsets {
-		// Skip any application that is neither in the conrol plane's namespace
+		// Skip any applicationsets that is neither in the conrol plane's namespace
 		// nor in the list of enabled namespaces.
 		if !security.IsNamespaceEnabled(a.Namespace, s.ns, s.enabledNamespaces) {
 			continue

server/applicationset/applicationset.proto

@@ -28,6 +28,15 @@ message ApplicationSetListQuery {
 	string appsetNamespace = 3;
 }
 
+message ApplicationSetWatchQuery {
+	string name = 1;
+	repeated string projects = 2;
+	string selector = 3;
+	string appSetNamespace = 4;
+	// when specified with a watch call, shows changes that occur after that particular version of a resource.
+	string resourceVersion = 5;
+}
+
 
 message ApplicationSetResponse {
 	string project = 1;
@@ -108,4 +117,8 @@ service ApplicationSetService {
 		option (google.api.http).get = "/api/v1/applicationsets/{name}/events";
 	}
 
+	rpc Watch (ApplicationSetWatchQuery) returns (stream github.com.argoproj.argo_cd.v3.pkg.apis.application.v1alpha1.ApplicationSetWatchEvent) {
+		option (google.api.http).get = "/api/v1/stream/applicationsets";
+	}
+
 }

server/applicationset/applicationset_test.go

@@ -201,8 +201,9 @@ func newTestAppSetServerWithEnforcerConfigure(t *testing.T, f func(*rbac.Enforce
 		enforcer,
 		nil,
 		fakeAppsClientset,
-		appInformer,
+		appsetInformer,
 		factory.Argoproj().V1alpha1().ApplicationSets().Lister(),
+		nil,
 		testNamespace,
 		sync.NewKeyLock(),
 		[]string{testNamespace, "external-namespace"},

server/broadcast/broadcaster.go

@@ -0,0 +1,117 @@
+package broadcast
+
+import (
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/watch"
+)
+
+// EventFunc creates a watch event from an object and event type.
+// T is the resource type (e.g., Application), E is the event type (e.g., ApplicationWatchEvent).
+type EventFunc[T any, E any] func(obj *T, eventType watch.EventType) *E
+
+// LogFieldsFunc returns log fields for an object (for logging dropped events)
+type LogFieldsFunc[T any] func(obj *T) log.Fields
+
+type subscriber[E any] struct {
+	ch      chan *E
+	filters []func(event *E) bool
+}
+
+func (s *subscriber[E]) matches(event *E) bool {
+	for i := range s.filters {
+		if !s.filters[i](event) {
+			return false
+		}
+	}
+	return true
+}
+
+// Broadcaster is an interface for broadcasting informer watch events to multiple subscribers.
+// T is the resource type (e.g., Application), E is the event type (e.g., ApplicationWatchEvent).
+type Broadcaster[E any] interface {
+	Subscribe(ch chan *E, filters ...func(event *E) bool) func()
+	OnAdd(any, bool)
+	OnUpdate(any, any)
+	OnDelete(any)
+}
+
+// Handler is a generic broadcaster handler that can be used for any resource type.
+// T is the resource type (e.g., Application), E is the event type (e.g., ApplicationWatchEvent).
+type Handler[T any, E any] struct {
+	lock        sync.Mutex
+	subscribers []*subscriber[E]
+	eventFunc   EventFunc[T, E]
+	logFields   LogFieldsFunc[T]
+}
+
+// NewHandler creates a new generic broadcaster handler.
+// T is the resource type (e.g., Application), E is the event type (e.g., ApplicationWatchEvent).
+func NewHandler[T any, E any](eventFunc EventFunc[T, E], logFields LogFieldsFunc[T]) *Handler[T, E] {
+	return &Handler[T, E]{
+		eventFunc: eventFunc,
+		logFields: logFields,
+	}
+}
+
+func (b *Handler[T, E]) notify(event *E, obj *T) {
+	// Make a local copy of b.subscribers, then send channel events outside the lock,
+	// to avoid data race on b.subscribers changes
+	var subscribers []*subscriber[E]
+	b.lock.Lock()
+	subscribers = append(subscribers, b.subscribers...)
+	b.lock.Unlock()
+
+	for _, s := range subscribers {
+		if s.matches(event) {
+			select {
+			case s.ch <- event:
+			default:
+				// drop event if cannot send right away
+				log.WithFields(b.logFields(obj)).Warn("unable to send event notification")
+			}
+		}
+	}
+}
+
+// Subscribe forwards informer watch events to the provided channel.
+// The watch events are dropped if no receivers are reading events from the channel so the channel must have
+// buffer if dropping events is not acceptable.
+func (b *Handler[T, E]) Subscribe(ch chan *E, filters ...func(event *E) bool) func() {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	sub := &subscriber[E]{ch, filters}
+	b.subscribers = append(b.subscribers, sub)
+	return func() {
+		b.lock.Lock()
+		defer b.lock.Unlock()
+		for i := range b.subscribers {
+			if b.subscribers[i] == sub {
+				b.subscribers = append(b.subscribers[:i], b.subscribers[i+1:]...)
+				break
+			}
+		}
+	}
+}
+
+func (b *Handler[T, E]) OnAdd(obj any, _ bool) {
+	if typedObj, ok := obj.(*T); ok {
+		event := b.eventFunc(typedObj, watch.Added)
+		b.notify(event, typedObj)
+	}
+}
+
+func (b *Handler[T, E]) OnUpdate(_, newObj any) {
+	if typedObj, ok := newObj.(*T); ok {
+		event := b.eventFunc(typedObj, watch.Modified)
+		b.notify(event, typedObj)
+	}
+}
+
+func (b *Handler[T, E]) OnDelete(obj any) {
+	if typedObj, ok := obj.(*T); ok {
+		event := b.eventFunc(typedObj, watch.Deleted)
+		b.notify(event, typedObj)
+	}
+}

server/broadcast/broadcaster_test.go

@@ -1,16 +1,25 @@
-package application
+package broadcast
 
 import (
 	"testing"
 	"time"
 
+	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/watch"
 
 	appv1 "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
 )
 
 func TestBroadcasterHandler_SubscribeUnsubscribe(t *testing.T) {
-	broadcaster := broadcasterHandler{}
+	broadcaster := NewHandler[appv1.Application, appv1.ApplicationWatchEvent](
+		func(app *appv1.Application, eventType watch.EventType) *appv1.ApplicationWatchEvent {
+			return &appv1.ApplicationWatchEvent{Application: *app, Type: eventType}
+		},
+		func(app *appv1.Application) log.Fields {
+			return log.Fields{"application": app.Name}
+		},
+	)
 
 	subscriber := make(chan *appv1.ApplicationWatchEvent)
 	unsubscribe := broadcaster.Subscribe(subscriber)
@@ -22,7 +31,14 @@ func TestBroadcasterHandler_SubscribeUnsubscribe(t *testing.T) {
 }
 
 func TestBroadcasterHandler_ReceiveEvents(t *testing.T) {
-	broadcaster := broadcasterHandler{}
+	broadcaster := NewHandler[appv1.Application, appv1.ApplicationWatchEvent](
+		func(app *appv1.Application, eventType watch.EventType) *appv1.ApplicationWatchEvent {
+			return &appv1.ApplicationWatchEvent{Application: *app, Type: eventType}
+		},
+		func(app *appv1.Application) log.Fields {
+			return log.Fields{"application": app.Name}
+		},
+	)
 
 	subscriber1 := make(chan *appv1.ApplicationWatchEvent, 1000)
 	subscriber2 := make(chan *appv1.ApplicationWatchEvent, 1000)
@@ -33,7 +49,7 @@ func TestBroadcasterHandler_ReceiveEvents(t *testing.T) {
 	firstReceived := false
 	secondReceived := false
 
-	go broadcaster.notify(&appv1.ApplicationWatchEvent{})
+	go broadcaster.OnAdd(&appv1.Application{}, false)
 
 	for {
 		select {

server/broadcast/mocks/Broadcaster.go

@@ -0,0 +1,240 @@
+// Code generated by mockery; DO NOT EDIT.
+// github.com/vektra/mockery
+// template: testify
+
+package mocks
+
+import (
+	mock "github.com/stretchr/testify/mock"
+)
+
+// NewBroadcaster creates a new instance of Broadcaster. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewBroadcaster[E any](t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *Broadcaster[E] {
+	mock := &Broadcaster[E]{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
+
+// Broadcaster is an autogenerated mock type for the Broadcaster type
+type Broadcaster[E any] struct {
+	mock.Mock
+}
+
+type Broadcaster_Expecter[E any] struct {
+	mock *mock.Mock
+}
+
+func (_m *Broadcaster[E]) EXPECT() *Broadcaster_Expecter[E] {
+	return &Broadcaster_Expecter[E]{mock: &_m.Mock}
+}
+
+// OnAdd provides a mock function for the type Broadcaster
+func (_mock *Broadcaster[E]) OnAdd(v any, b bool) {
+	_mock.Called(v, b)
+	return
+}
+
+// Broadcaster_OnAdd_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnAdd'
+type Broadcaster_OnAdd_Call[E any] struct {
+	*mock.Call
+}
+
+// OnAdd is a helper method to define mock.On call
+//   - v any
+//   - b bool
+func (_e *Broadcaster_Expecter[E]) OnAdd(v interface{}, b interface{}) *Broadcaster_OnAdd_Call[E] {
+	return &Broadcaster_OnAdd_Call[E]{Call: _e.mock.On("OnAdd", v, b)}
+}
+
+func (_c *Broadcaster_OnAdd_Call[E]) Run(run func(v any, b bool)) *Broadcaster_OnAdd_Call[E] {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 any
+		if args[0] != nil {
+			arg0 = args[0].(any)
+		}
+		var arg1 bool
+		if args[1] != nil {
+			arg1 = args[1].(bool)
+		}
+		run(
+			arg0,
+			arg1,
+		)
+	})
+	return _c
+}
+
+func (_c *Broadcaster_OnAdd_Call[E]) Return() *Broadcaster_OnAdd_Call[E] {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *Broadcaster_OnAdd_Call[E]) RunAndReturn(run func(v any, b bool)) *Broadcaster_OnAdd_Call[E] {
+	_c.Run(run)
+	return _c
+}
+
+// OnDelete provides a mock function for the type Broadcaster
+func (_mock *Broadcaster[E]) OnDelete(v any) {
+	_mock.Called(v)
+	return
+}
+
+// Broadcaster_OnDelete_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnDelete'
+type Broadcaster_OnDelete_Call[E any] struct {
+	*mock.Call
+}
+
+// OnDelete is a helper method to define mock.On call
+//   - v any
+func (_e *Broadcaster_Expecter[E]) OnDelete(v interface{}) *Broadcaster_OnDelete_Call[E] {
+	return &Broadcaster_OnDelete_Call[E]{Call: _e.mock.On("OnDelete", v)}
+}
+
+func (_c *Broadcaster_OnDelete_Call[E]) Run(run func(v any)) *Broadcaster_OnDelete_Call[E] {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 any
+		if args[0] != nil {
+			arg0 = args[0].(any)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *Broadcaster_OnDelete_Call[E]) Return() *Broadcaster_OnDelete_Call[E] {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *Broadcaster_OnDelete_Call[E]) RunAndReturn(run func(v any)) *Broadcaster_OnDelete_Call[E] {
+	_c.Run(run)
+	return _c
+}
+
+// OnUpdate provides a mock function for the type Broadcaster
+func (_mock *Broadcaster[E]) OnUpdate(v any, v1 any) {
+	_mock.Called(v, v1)
+	return
+}
+
+// Broadcaster_OnUpdate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'OnUpdate'
+type Broadcaster_OnUpdate_Call[E any] struct {
+	*mock.Call
+}
+
+// OnUpdate is a helper method to define mock.On call
+//   - v any
+//   - v1 any
+func (_e *Broadcaster_Expecter[E]) OnUpdate(v interface{}, v1 interface{}) *Broadcaster_OnUpdate_Call[E] {
+	return &Broadcaster_OnUpdate_Call[E]{Call: _e.mock.On("OnUpdate", v, v1)}
+}
+
+func (_c *Broadcaster_OnUpdate_Call[E]) Run(run func(v any, v1 any)) *Broadcaster_OnUpdate_Call[E] {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 any
+		if args[0] != nil {
+			arg0 = args[0].(any)
+		}
+		var arg1 any
+		if args[1] != nil {
+			arg1 = args[1].(any)
+		}
+		run(
+			arg0,
+			arg1,
+		)
+	})
+	return _c
+}
+
+func (_c *Broadcaster_OnUpdate_Call[E]) Return() *Broadcaster_OnUpdate_Call[E] {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *Broadcaster_OnUpdate_Call[E]) RunAndReturn(run func(v any, v1 any)) *Broadcaster_OnUpdate_Call[E] {
+	_c.Run(run)
+	return _c
+}
+
+// Subscribe provides a mock function for the type Broadcaster
+func (_mock *Broadcaster[E]) Subscribe(ch chan *E, filters ...func(event *E) bool) func() {
+	// func(event *E) bool
+	_va := make([]interface{}, len(filters))
+	for _i := range filters {
+		_va[_i] = filters[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ch)
+	_ca = append(_ca, _va...)
+	ret := _mock.Called(_ca...)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Subscribe")
+	}
+
+	var r0 func()
+	if returnFunc, ok := ret.Get(0).(func(chan *E, ...func(event *E) bool) func()); ok {
+		r0 = returnFunc(ch, filters...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(func())
+		}
+	}
+	return r0
+}
+
+// Broadcaster_Subscribe_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Subscribe'
+type Broadcaster_Subscribe_Call[E any] struct {
+	*mock.Call
+}
+
+// Subscribe is a helper method to define mock.On call
+//   - ch chan *E
+//   - filters ...func(event *E) bool
+func (_e *Broadcaster_Expecter[E]) Subscribe(ch interface{}, filters ...interface{}) *Broadcaster_Subscribe_Call[E] {
+	return &Broadcaster_Subscribe_Call[E]{Call: _e.mock.On("Subscribe",
+		append([]interface{}{ch}, filters...)...)}
+}
+
+func (_c *Broadcaster_Subscribe_Call[E]) Run(run func(ch chan *E, filters ...func(event *E) bool)) *Broadcaster_Subscribe_Call[E] {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 chan *E
+		if args[0] != nil {
+			arg0 = args[0].(chan *E)
+		}
+		var arg1 []func(event *E) bool
+		variadicArgs := make([]func(event *E) bool, len(args)-1)
+		for i, a := range args[1:] {
+			if a != nil {
+				variadicArgs[i] = a.(func(event *E) bool)
+			}
+		}
+		arg1 = variadicArgs
+		run(
+			arg0,
+			arg1...,
+		)
+	})
+	return _c
+}
+
+func (_c *Broadcaster_Subscribe_Call[E]) Return(fn func()) *Broadcaster_Subscribe_Call[E] {
+	_c.Call.Return(fn)
+	return _c
+}
+
+func (_c *Broadcaster_Subscribe_Call[E]) RunAndReturn(run func(ch chan *E, filters ...func(event *E) bool) func()) *Broadcaster_Subscribe_Call[E] {
+	_c.Call.Return(run)
+	return _c
+}

server/server.go

@@ -1061,6 +1061,7 @@ func newArgoCDServiceSet(a *ArgoCDServer) *ArgoCDServiceSet {
 		a.AppClientset,
 		a.appsetInformer,
 		a.appsetLister,
+		nil,
 		a.Namespace,
 		projectLock,
 		a.ApplicationNamespaces,

util/app/log/log.go

@@ -13,3 +13,11 @@ func GetAppLogFields(app *appv1.Application) log.Fields {
 		"project":       app.Spec.Project,
 	}
 }
+
+func GetAppSetLogFields(appset *appv1.ApplicationSet) log.Fields {
+	return log.Fields{
+		"applicationSet":   appset.Name,
+		"appSet-namespace": appset.Namespace,
+		"project":          appset.Spec.Template.Spec.Project,
+	}
+}