APIGOV-33163 - new event kind handler structure

.gitignore

@@ -41,3 +41,4 @@ sub-resources.json
 vendor
 sample.sequence
 **/data/cache
+.claude

pkg/agent/agent.go

@@ -803,52 +803,66 @@ func cleanUp() {
 	}
 }
 
-func newHandlers() []handler.Handler {
+// kindsProvider is implemented by Handlers whose Kind set isn't known statically at the call
+// sites below - it's derived from config (watchResourceHandler) or from target handlers
+// registered elsewhere via RegisterProvisioner/RegisterResourceEventHandler (proxyResourceHandler).
+type kindsProvider interface {
+	Kinds() []string
+}
+
+func newHandlers() map[string][]handler.Handler {
 	envName := GetCentralConfig().GetEnvironmentName()
-	handlers := []handler.Handler{
-		handler.NewAPISvcHandler(agent.cacheManager, envName),
-		handler.NewInstanceHandler(agent.cacheManager, envName),
-		handler.NewAgentResourceHandler(agent.agentResourceManager, sampling.GetGlobalSampling(), agent.cacheManager, agent.apicClient),
-		agent.proxyResourceHandler,
+
+	handlers := map[string][]handler.Handler{}
+	addByKindsProvider := func(h handler.Handler) {
+		kp, ok := h.(kindsProvider)
+		if !ok {
+			return
+		}
+		for _, kind := range kp.Kinds() {
+			handlers[kind] = append(handlers[kind], h)
+		}
 	}
 
 	switch agent.cfg.GetAgentType() {
 	case config.DiscoveryAgent:
-		handlers = append(
-			handlers,
-			handler.NewWatchResourceHandler(agent.cacheManager, handler.WithWatchTopicFeatures(agent.cfg)),
-			handler.NewCRDHandler(agent.cacheManager),
-			handler.NewARDHandler(agent.cacheManager),
-			handler.NewAPDHandler(agent.cacheManager),
-			handler.NewEnvironmentHandler(agent.cacheManager, agent.cfg.GetCredentialConfig(), envName),
-			handler.NewDiscoveryManagedApplicationHandler(agent.cacheManager),
-			handler.NewDiscoveryAccessRequestHandler(agent.cacheManager),
-			handler.NewIDPHandler(agent.cacheManager, agent.cfg.GetCredentialConfig()),
-		)
+		addByKindsProvider(handler.NewWatchResourceHandler(agent.cacheManager, handler.WithWatchTopicFeatures(agent.cfg)))
+		handlers[management.CredentialRequestDefinitionGVK().Kind] = append(handlers[management.CredentialRequestDefinitionGVK().Kind], handler.NewCRDHandler(agent.cacheManager))
+		handlers[management.AccessRequestDefinitionGVK().Kind] = append(handlers[management.AccessRequestDefinitionGVK().Kind], handler.NewARDHandler(agent.cacheManager))
+		handlers[management.ApplicationProfileDefinitionGVK().Kind] = append(handlers[management.ApplicationProfileDefinitionGVK().Kind], handler.NewAPDHandler(agent.cacheManager))
+		handlers[management.EnvironmentGVK().Kind] = append(handlers[management.EnvironmentGVK().Kind], handler.NewEnvironmentHandler(agent.cacheManager, agent.cfg.GetCredentialConfig(), envName))
+		handlers[management.ManagedApplicationGVK().Kind] = append(handlers[management.ManagedApplicationGVK().Kind], handler.NewManagedApplicationCacheHandler(agent.cacheManager))
+		handlers[management.AccessRequestGVK().Kind] = append(handlers[management.AccessRequestGVK().Kind], handler.NewAccessRequestCacheHandler(config.DiscoveryAgent, agent.cacheManager, nil))
+		handlers[management.IdentityProviderMetadataGVK().Kind] = append(handlers[management.IdentityProviderMetadataGVK().Kind], handler.NewIDPHandler(agent.cacheManager, agent.cfg.GetCredentialConfig()))
 	case config.TraceabilityAgent:
 		// Register managed application and access handler for traceability agent
 		// For discovery agent, the handlers gets registered while setting up provisioner
-		handlers = append(
-			handlers,
-			handler.NewWatchResourceHandler(agent.cacheManager, handler.WithWatchTopicFeatures(agent.cfg)),
-			handler.NewTraceAccessRequestHandler(agent.cacheManager, agent.apicClient),
-			handler.NewTraceManagedApplicationHandler(agent.cacheManager),
-		)
+		addByKindsProvider(handler.NewWatchResourceHandler(agent.cacheManager, handler.WithWatchTopicFeatures(agent.cfg)))
+		handlers[management.AccessRequestGVK().Kind] = append(handlers[management.AccessRequestGVK().Kind], handler.NewAccessRequestCacheHandler(config.TraceabilityAgent, agent.cacheManager, agent.apicClient))
+		handlers[management.ManagedApplicationGVK().Kind] = append(handlers[management.ManagedApplicationGVK().Kind], handler.NewManagedApplicationCacheHandler(agent.cacheManager))
 	case config.ComplianceAgent:
-		handlers = append(
-			handlers,
-			handler.NewWatchResourceHandler(agent.cacheManager,
-				handler.WithWatchTopicFeatures(agent.cfg),
-				handler.WithWatchTopicGroupKind(
-					[]apiV1.GroupKind{
-						management.EnvironmentGVK().GroupKind,
-						management.APIServiceInstanceGVK().GroupKind,
-						management.ComplianceRuntimeResultGVK().GroupKind,
-					},
-				),
+		addByKindsProvider(handler.NewWatchResourceHandler(agent.cacheManager,
+			handler.WithWatchTopicFeatures(agent.cfg),
+			handler.WithWatchTopicGroupKind(
+				[]apiV1.GroupKind{
+					management.EnvironmentGVK().GroupKind,
+					management.APIServiceInstanceGVK().GroupKind,
+					management.ComplianceRuntimeResultGVK().GroupKind,
+				},
 			),
-			handler.NewCRRHandler(agent.cacheManager),
-		)
+		))
+		handlers[management.ComplianceRuntimeResultGVK().Kind] = append(handlers[management.ComplianceRuntimeResultGVK().Kind], handler.NewCRRHandler(agent.cacheManager))
+	}
+
+	handlers[management.APIServiceGVK().Kind] = append(handlers[management.APIServiceGVK().Kind], handler.NewAPISvcHandler(agent.cacheManager, envName))
+	handlers[management.APIServiceInstanceGVK().Kind] = append(handlers[management.APIServiceInstanceGVK().Kind], handler.NewInstanceHandler(agent.cacheManager, envName))
+
+	agentResHandler := handler.NewAgentResourceHandler(agent.agentResourceManager, sampling.GetGlobalSampling(), agent.cacheManager, agent.apicClient)
+	handlers[management.DiscoveryAgentGVK().Kind] = append(handlers[management.DiscoveryAgentGVK().Kind], agentResHandler)
+	handlers[management.TraceabilityAgentGVK().Kind] = append(handlers[management.TraceabilityAgentGVK().Kind], agentResHandler)
+	handlers[management.ComplianceAgentGVK().Kind] = append(handlers[management.ComplianceAgentGVK().Kind], agentResHandler)
+	for kind, proxyHandler := range agent.proxyResourceHandler.GetHandlers() {
+		handlers[kind] = append(handlers[kind], proxyHandler)
 	}
 
 	return handlers

pkg/agent/discoverycache.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/Axway/agent-sdk/pkg/agent/handler"
 	"github.com/Axway/agent-sdk/pkg/migrate"
-	"github.com/Axway/agent-sdk/pkg/watchmanager/proto"
 
 	apiv1 "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/api/v1"
 	management "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/management/v1"
@@ -19,7 +18,7 @@ type discoveryCache struct {
 	centralURL               string
 	migrator                 migrate.Migrator
 	logger                   log.FieldLogger
-	handlers                 []handler.Handler
+	handlersByKind           map[string][]handler.Handler
 	client                   resourceClient
 	additionalDiscoveryFuncs []discoverFunc
 	watchTopic               *management.WatchTopic
@@ -59,7 +58,7 @@ func preMarketplaceSetup(f func() error) discoveryOpt {
 func newDiscoveryCache(
 	cfg config.CentralConfig,
 	client resourceClient,
-	handlers []handler.Handler,
+	handlersByKind map[string][]handler.Handler,
 	watchTopic *management.WatchTopic,
 	opts ...discoveryOpt,
 ) *discoveryCache {
@@ -69,7 +68,7 @@ func newDiscoveryCache(
 
 	dc := &discoveryCache{
 		logger:                   logger,
-		handlers:                 handlers,
+		handlersByKind:           handlersByKind,
 		centralURL:               cfg.GetURL(),
 		client:                   client,
 		additionalDiscoveryFuncs: make([]discoverFunc, 0),
@@ -290,8 +289,7 @@ func (dc *discoveryCache) handleResourcesList(list []*apiv1.ResourceInstance) er
 			}
 		}
 
-		action := getAction(ri.Metadata.State)
-		if err := dc.handleResource(ri, action); err != nil {
+		if err := dc.handleResource(ri); err != nil {
 			logger.
 				WithError(err).
 				Error("failed to handle resource")
@@ -301,24 +299,19 @@ func (dc *discoveryCache) handleResourcesList(list []*apiv1.ResourceInstance) er
 	return nil
 }
 
-func (dc *discoveryCache) handleResource(ri *apiv1.ResourceInstance, action proto.Event_Type) error {
-	ctx := handler.NewEventContext(action, nil, ri.Name, ri.Kind)
-	for _, h := range dc.handlers {
-		err := h.Handle(ctx, nil, ri)
-		if err != nil {
+func (dc *discoveryCache) handleResource(ri *apiv1.ResourceInstance) error {
+	for _, h := range dc.handlersByKind[ri.Kind] {
+		ch, ok := h.(handler.CacheHandler)
+		if !ok {
+			continue
+		}
+		if err := ch.HandleCache(ri); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func getAction(state string) proto.Event_Type {
-	if state == apiv1.ResourceDeleting {
-		return proto.Event_UPDATED
-	}
-	return proto.Event_CREATED
-}
-
 func isMPResource(kind string) bool {
 	switch kind {
 	case management.ManagedApplicationGVK().Kind:

pkg/agent/discoverycache_test.go

@@ -99,7 +99,13 @@ func TestDiscoveryCacheExecute(t *testing.T) {
 			}
 
 			dc := newDiscoveryCache(cfg, c,
-				[]handler.Handler{svcHandler, managedAppHandler, managedAppProfHandler, accessReqHandler, credHandler},
+				map[string][]handler.Handler{
+					management.APIServiceGVK().Kind:               {svcHandler},
+					management.ManagedApplicationGVK().Kind:        {managedAppHandler},
+					management.ManagedApplicationProfileGVK().Kind: {managedAppProfHandler},
+					management.AccessRequestGVK().Kind:             {accessReqHandler},
+					management.CredentialGVK().Kind:                {credHandler},
+				},
 				tc.wt,
 				opts...,
 			)
@@ -143,6 +149,18 @@ func (m *mockHandler) Handle(_ context.Context, _ *proto.EventMeta, ri *apiv1.Re
 	return m.err
 }
 
+func (m *mockHandler) ShouldHandle(_ context.Context, _ *proto.Event) bool {
+	return true
+}
+
+func (m *mockHandler) HandleCache(ri *apiv1.ResourceInstance) error {
+	if m.kind != "" && ri.Kind != m.kind {
+		return nil
+	}
+	m.count = m.count + 1
+	return m.err
+}
+
 func newAPIServices(scope string) []*apiv1.ResourceInstance {
 	svc1, _ := management.NewAPIService("svc1", scope).AsInstance()
 	svc2, _ := management.NewAPIService("svc2", scope).AsInstance()

pkg/agent/events/eventlistener.go

@@ -30,17 +30,18 @@ type EventListener struct {
 	ctx             context.Context
 	cancel          context.CancelCauseFunc
 	client          APIClient
-	handlers        []handler.Handler
+	handlersByKind  map[string][]handler.Handler
 	logger          log.FieldLogger
 	sequenceManager SequenceProvider
 	source          chan *proto.Event
 }
 
 // NewListenerFunc type for creating a new listener
-type NewListenerFunc func(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener
+type NewListenerFunc func(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, handlersByKind map[string][]handler.Handler) *EventListener
 
-// NewEventListener creates a new EventListener to process events based on the provided Handlers.
-func NewEventListener(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, cbs ...handler.Handler) *EventListener {
+// NewEventListener creates a new EventListener to process events based on the provided Handlers,
+// indexed by the resource Kind they should be dispatched for.
+func NewEventListener(ctx context.Context, cancel context.CancelCauseFunc, source chan *proto.Event, client APIClient, sequenceManager SequenceProvider, handlersByKind map[string][]handler.Handler) *EventListener {
 	logger := log.NewFieldLogger().
 		WithComponent("EventListener").
 		WithPackage("sdk.agent.events")
@@ -49,7 +50,7 @@ func NewEventListener(ctx context.Context, cancel context.CancelCauseFunc, sourc
 		ctx:             ctx,
 		cancel:          cancel,
 		client:          client,
-		handlers:        cbs,
+		handlersByKind:  handlersByKind,
 		logger:          logger,
 		sequenceManager: sequenceManager,
 		source:          source,
@@ -112,14 +113,26 @@ func (em *EventListener) handleEvent(event *proto.Event) error {
 		WithField("kind", event.Payload.Kind).
 		WithField("name", event.Payload.Name).
 		WithField("type", event.Type.String()).
+		WithField("subResource", event.Metadata.Subresource).
 		Debug("processing watch event")
 
-	ri, err := em.getEventResource(event)
-	if err != nil {
-		return err
+	var ri *apiv1.ResourceInstance
+	var err error
+	for _, h := range em.handlersByKind[event.Payload.Kind] {
+		if !h.ShouldHandle(ctx, event) {
+			continue
+		}
+		if ri == nil {
+			ri, err = em.getEventResource(event)
+			if err != nil {
+				return err
+			}
+		}
+		if err := h.Handle(ctx, event.Metadata, ri); err != nil {
+			em.logger.Error(err)
+		}
 	}
 
-	em.HandleResource(ctx, event.Metadata, ri)
 	em.sequenceManager.SetSequence(event.Metadata.SequenceID)
 	return nil
 }
@@ -131,20 +144,6 @@ func (em *EventListener) getEventResource(event *proto.Event) (*apiv1.ResourceIn
 	return em.client.GetResource(event.Payload.Metadata.SelfLink)
 }
 
-// HandleResource loops through all the handlers and passes the event to each one for processing.
-func (em *EventListener) HandleResource(
-	ctx context.Context,
-	eventMetadata *proto.EventMeta,
-	resource *apiv1.ResourceInstance,
-) {
-	for _, h := range em.handlers {
-		err := h.Handle(ctx, eventMetadata, resource)
-		if err != nil {
-			em.logger.Error(err)
-		}
-	}
-}
-
 func (em *EventListener) convertEventPayload(event *proto.Event) *apiv1.ResourceInstance {
 	ri := &apiv1.ResourceInstance{
 		ResourceMeta: apiv1.ResourceMeta{

pkg/agent/events/eventlistener_test.go

@@ -62,7 +62,7 @@ func TestEventListener_start(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, cancel := context.WithCancelCause(context.Background())
-			listener := NewEventListener(ctx, cancel, tc.events, tc.client, sequenceManager, tc.handler)
+			listener := NewEventListener(ctx, cancel, tc.events, tc.client, sequenceManager, map[string][]handler.Handler{"": {tc.handler}})
 
 			errCh := make(chan error)
 			go func() {
@@ -103,14 +103,14 @@ func TestEventListener_Listen(t *testing.T) {
 	sequenceManager := NewSequenceProvider(cacheManager, "testWatch")
 	events := make(chan *proto.Event)
 	ctx, cancel := context.WithCancelCause(context.Background())
-	listener := NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, &mockHandler{})
+	listener := NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, map[string][]handler.Handler{"": {&mockHandler{}}})
 	listener.Listen()
 	listener.Stop()
 	err := ctx.Err()
 	assert.NotNil(t, err)
 
 	ctx, cancel = context.WithCancelCause(context.Background())
-	listener = NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, &mockHandler{})
+	listener = NewEventListener(ctx, cancel, events, &mockAPIClient{}, sequenceManager, map[string][]handler.Handler{"": {&mockHandler{}}})
 	listener.Listen()
 	close(events)
 	err = ctx.Err()
@@ -176,7 +176,7 @@ func TestEventListener_handleEvent(t *testing.T) {
 			}
 
 			ctx, cancel := context.WithCancelCause(context.Background())
-			listener := NewEventListener(ctx, cancel, make(chan *proto.Event), tc.client, sequenceManager, tc.handler)
+			listener := NewEventListener(ctx, cancel, make(chan *proto.Event), tc.client, sequenceManager, map[string][]handler.Handler{"": {tc.handler}})
 
 			err := listener.handleEvent(event)
 
@@ -197,6 +197,10 @@ func (m *mockHandler) Handle(_ context.Context, _ *proto.EventMeta, _ *apiv1.Res
 	return m.err
 }
 
+func (m *mockHandler) ShouldHandle(_ context.Context, _ *proto.Event) bool {
+	return true
+}
+
 // slowHandler blocks for the given duration each time Handle is called,
 // and atomically increments callCount.
 type slowHandler struct {
@@ -210,6 +214,10 @@ func (h *slowHandler) Handle(_ context.Context, _ *proto.EventMeta, _ *apiv1.Res
 	return nil
 }
 
+func (h *slowHandler) ShouldHandle(_ context.Context, _ *proto.Event) bool {
+	return true
+}
+
 func newTestEvent(seqID int64) *proto.Event {
 	return &proto.Event{
 		Type: proto.Event_CREATED,

pkg/agent/eventsync_test.go

@@ -175,7 +175,7 @@ func TestEventSync(t *testing.T) {
 
 				svcHandler := &mockHandler{kind: management.APIServiceGVK().Kind}
 				wt := &management.WatchTopic{Spec: management.WatchTopicSpec{Filters: tc.wtFilters}}
-				dc := newDiscoveryCache(cfg, tc.makeClient(), []handler.Handler{svcHandler}, wt)
+				dc := newDiscoveryCache(cfg, tc.makeClient(), map[string][]handler.Handler{management.APIServiceGVK().Kind: {svcHandler}}, wt)
 
 				es := &EventSync{
 					watchTopic:     wt,