Add agentGateway injection options (#12999)

Signed-off-by: David Jumani <[email protected]>

Author: davidjumani

pkg/agentgateway/translator/gateway_collection.go

@@ -194,31 +194,62 @@ func (g ParentInfo) Equals(other ParentInfo) bool {
 		slices.Equal(g.Hostnames, other.Hostnames)
 }
 
+type GatewayTransformationFunction func(GatewayCollectionConfig) func(ctx krt.HandlerContext, obj *gwv1.Gateway) (*gwv1.GatewayStatus, []*GatewayListener)
+
+type GatewayCollectionConfigOption func(o *GatewayCollectionConfig)
+
+func WithGatewayTransformationFunc(f GatewayTransformationFunction) GatewayCollectionConfigOption {
+	return func(o *GatewayCollectionConfig) {
+		o.transformationFunc = f
+	}
+}
+
+type GatewayCollectionConfig struct {
+	ControllerName string
+	Gateways       krt.Collection[*gwv1.Gateway]
+	ListenerSets   krt.Collection[ListenerSet]
+	GatewayClasses krt.Collection[GatewayClass]
+	Namespaces     krt.Collection[*corev1.Namespace]
+	Grants         ReferenceGrants
+	Secrets        krt.Collection[*corev1.Secret]
+	ConfigMaps     krt.Collection[*corev1.ConfigMap]
+	KrtOpts        krtutil.KrtOptions
+
+	listenerIndex      krt.Index[types.NamespacedName, ListenerSet]
+	transformationFunc GatewayTransformationFunction
+}
+
 // GatewayCollection returns a collection of the internal representations GatewayListeners for the given gateway.
 func GatewayCollection(
-	controllerName string,
-	gateways krt.Collection[*gwv1.Gateway],
-	listenerSets krt.Collection[ListenerSet],
-	gatewayClasses krt.Collection[GatewayClass],
-	namespaces krt.Collection[*corev1.Namespace],
-	grants ReferenceGrants,
-	secrets krt.Collection[*corev1.Secret],
-	configMaps krt.Collection[*corev1.ConfigMap],
-	krtopts krtutil.KrtOptions,
+	cfg GatewayCollectionConfig,
+	opts ...GatewayCollectionConfigOption,
 ) (
 	krt.StatusCollection[*gwv1.Gateway, gwv1.GatewayStatus],
 	krt.Collection[*GatewayListener],
 ) {
-	listenerIndex := krt.NewIndex(listenerSets, "gatewayParent", func(o ListenerSet) []types.NamespacedName {
+	for _, fn := range opts {
+		fn(&cfg)
+	}
+	cfg.listenerIndex = krt.NewIndex(cfg.ListenerSets, "gatewayParent", func(o ListenerSet) []types.NamespacedName {
 		return []types.NamespacedName{o.GatewayParent}
 	})
-	statusCol, gw := krt.NewStatusManyCollection(gateways, func(ctx krt.HandlerContext, obj *gwv1.Gateway) (*gwv1.GatewayStatus, []*GatewayListener) {
-		class := krt.FetchOne(ctx, gatewayClasses, krt.FilterKey(string(obj.Spec.GatewayClassName)))
+	if cfg.transformationFunc == nil {
+		cfg.transformationFunc = GatewaysTransformationFunc
+	}
+
+	statusCol, gw := krt.NewStatusManyCollection(cfg.Gateways, cfg.transformationFunc(cfg), cfg.KrtOpts.ToOptions("KubernetesGateway")...)
+
+	return statusCol, gw
+}
+
+func GatewaysTransformationFunc(cfg GatewayCollectionConfig) func(ctx krt.HandlerContext, obj *gwv1.Gateway) (*gwv1.GatewayStatus, []*GatewayListener) {
+	return func(ctx krt.HandlerContext, obj *gwv1.Gateway) (*gwv1.GatewayStatus, []*GatewayListener) {
+		class := krt.FetchOne(ctx, cfg.GatewayClasses, krt.FilterKey(string(obj.Spec.GatewayClassName)))
 		if class == nil {
 			logger.Debug("gateway class not found, skipping", "gw_name", obj.GetName(), "gatewayClassName", obj.Spec.GatewayClassName)
 			return nil, nil
 		}
-		if string(class.Controller) != controllerName {
+		if string(class.Controller) != cfg.ControllerName {
 			logger.Debug("skipping gateway not managed by our controller", "gw_name", obj.GetName(), "gatewayClassName", obj.Spec.GatewayClassName, "controllerName", class.Controller)
 			return nil, nil // ignore gateways not managed by our controller
 		}
@@ -249,7 +280,7 @@ func GatewayCollection(
 			// Attached Routes count starts at 0 and gets updated later in the status syncer
 			// when the real count is available after route processing
 
-			hostnames, tlsInfo, updatedStatus, programmed := BuildListener(ctx, secrets, configMaps, grants, namespaces, obj, status.Listeners, kgw, l, i, nil)
+			hostnames, tlsInfo, updatedStatus, programmed := BuildListener(ctx, cfg.Secrets, cfg.ConfigMaps, cfg.Grants, cfg.Namespaces, obj, status.Listeners, kgw, l, i, nil)
 			status.Listeners = updatedStatus
 
 			lstatus := status.Listeners[i]
@@ -303,7 +334,7 @@ func GatewayCollection(
 			})
 			result = append(result, res)
 		}
-		listenersFromSets := krt.Fetch(ctx, listenerSets, krt.FilterIndex(listenerIndex, config.NamespacedName(obj)))
+		listenersFromSets := krt.Fetch(ctx, cfg.ListenerSets, krt.FilterIndex(cfg.listenerIndex, config.NamespacedName(obj)))
 		for _, ls := range listenersFromSets {
 			result = append(result, &GatewayListener{
 				Name:          ls.Name,
@@ -320,9 +351,7 @@ func GatewayCollection(
 		}
 		gws := rm.BuildGWStatus(context.Background(), *obj, nil)
 		return gws, result
-	}, krtopts.ToOptions("KubernetesGateway")...)
-
-	return statusCol, gw
+	}
 }
 
 type ListenerSet struct {

pkg/kgateway/agentgatewaysyncer/option.go

@@ -0,0 +1,23 @@
+package agentgatewaysyncer
+
+import "github.com/kgateway-dev/kgateway/v2/pkg/agentgateway/translator"
+
+type agentgatewaySyncerConfig struct {
+	GatewayTransformationFunc translator.GatewayTransformationFunction
+}
+
+type AgentgatewaySyncerOption func(*agentgatewaySyncerConfig)
+
+func processAgentgatewaySyncerOptions(opts ...AgentgatewaySyncerOption) *agentgatewaySyncerConfig {
+	cfg := &agentgatewaySyncerConfig{}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return cfg
+}
+
+func WithGatewayTransformationFunc(f translator.GatewayTransformationFunction) AgentgatewaySyncerOption {
+	return func(o *agentgatewaySyncerConfig) {
+		o.GatewayTransformationFunc = f
+	}
+}

pkg/kgateway/agentgatewaysyncer/syncer.go

@@ -68,6 +68,8 @@ type Syncer struct {
 	Registrations []krtxds.Registration
 
 	Outputs OutputCollections
+
+	gatewayCollectionOptions []translator.GatewayCollectionConfigOption
 }
 
 func NewAgwSyncer(
@@ -79,7 +81,9 @@ func NewAgwSyncer(
 	additionalGatewayClasses map[string]*deployer.GatewayClassInfo,
 	krtopts krtutil.KrtOptions,
 	extraGVKs []schema.GroupVersionKind,
+	opts ...AgentgatewaySyncerOption,
 ) *Syncer {
+	cfg := processAgentgatewaySyncerOptions(opts...)
 	syncer := &Syncer{
 		agwCollections:           agwCollections,
 		controllerName:           controllerName,
@@ -89,6 +93,8 @@ func NewAgwSyncer(
 		client:                   client,
 		statusCollections:        status.NewStatusCollections(extraGVKs),
 		NackPublisher:            nack.NewPublisher(client),
+		gatewayCollectionOptions: []translator.GatewayCollectionConfigOption{
+			translator.WithGatewayTransformationFunc(cfg.GatewayTransformationFunc)},
 	}
 	logger.Debug("init agentgateway Syncer", "controllername", controllerName)
 
@@ -174,17 +180,17 @@ func (s *Syncer) buildGatewayCollection(
 	krt.StatusCollection[*gwv1.Gateway, gwv1.GatewayStatus],
 	krt.Collection[*translator.GatewayListener],
 ) {
-	return translator.GatewayCollection(
-		s.controllerName,
-		s.agwCollections.Gateways,
-		listenerSets,
-		gatewayClasses,
-		s.agwCollections.Namespaces,
-		refGrants,
-		s.agwCollections.Secrets,
-		s.agwCollections.ConfigMaps,
-		krtopts,
-	)
+	return translator.GatewayCollection(translator.GatewayCollectionConfig{
+		ControllerName: s.controllerName,
+		Gateways:       s.agwCollections.Gateways,
+		ListenerSets:   listenerSets,
+		GatewayClasses: gatewayClasses,
+		Namespaces:     s.agwCollections.Namespaces,
+		Grants:         refGrants,
+		Secrets:        s.agwCollections.Secrets,
+		ConfigMaps:     s.agwCollections.ConfigMaps,
+		KrtOpts:        krtopts,
+	}, s.gatewayCollectionOptions...)
 }
 
 func (s *Syncer) buildListenerSetCollection(

pkg/kgateway/controller/start.go

@@ -39,7 +39,6 @@ import (
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/ir"
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/krtutil"
 	kgtwschemes "github.com/kgateway-dev/kgateway/v2/pkg/schemes"
-	"github.com/kgateway-dev/kgateway/v2/pkg/syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/utils/kubeutils"
 	"github.com/kgateway-dev/kgateway/v2/pkg/utils/namespaces"
 	"github.com/kgateway-dev/kgateway/v2/pkg/validator"
@@ -101,7 +100,10 @@ type StartConfig struct {
 	GatewayControllerExtension sdk.GatewayControllerExtension
 
 	// StatusSyncerOptions is the list of options to be passed when creating the StatusSyncer
-	StatusSyncerOptions []syncer.StatusSyncerOption
+	StatusSyncerOptions []proxy_syncer.StatusSyncerOption
+
+	// AgentgatewaySyncerOptions is the list of options to be passed when creating the AgentGatewaySyncer
+	AgentgatewaySyncerOptions []agentgatewaysyncer.AgentgatewaySyncerOption
 }
 
 // Start runs the controllers responsible for processing the K8s Gateway API objects
@@ -227,6 +229,7 @@ func NewControllerBuilder(ctx context.Context, cfg StartConfig) (*ControllerBuil
 			cfg.AdditionalGatewayClasses,
 			cfg.KrtOptions,
 			gvks,
+			cfg.AgentgatewaySyncerOptions...,
 		)
 
 		if err := cfg.Manager.Add(agwSyncer); err != nil {

pkg/syncer/option.go pkg/kgateway/proxy_syncer/option.gopkg/syncer/option.go renamed to pkg/kgateway/proxy_syncer/option.go

@@ -1,27 +1,27 @@
-package syncer
+package proxy_syncer
 
 import (
 	"context"
 
 	"github.com/kgateway-dev/kgateway/v2/pkg/reports"
 )
 
-type StatusSyncerConfig struct {
+type statusSyncerConfig struct {
 	CustomStatusSync func(ctx context.Context, rm reports.ReportMap)
 }
 
-type StatusSyncerOption func(*StatusSyncerConfig)
+type StatusSyncerOption func(*statusSyncerConfig)
 
-func ProcessStatusSyncerOptions(opts ...StatusSyncerOption) *StatusSyncerConfig {
-	cfg := &StatusSyncerConfig{}
+func processStatusSyncerOptions(opts ...StatusSyncerOption) *statusSyncerConfig {
+	cfg := &statusSyncerConfig{}
 	for _, fn := range opts {
 		fn(cfg)
 	}
 	return cfg
 }
 
 func WithCustomStatusSync(customSync func(ctx context.Context, rm reports.ReportMap)) StatusSyncerOption {
-	return func(cfg *StatusSyncerConfig) {
+	return func(cfg *statusSyncerConfig) {
 		cfg.CustomStatusSync = customSync
 	}
 }

pkg/kgateway/proxy_syncer/status_syncer.go

@@ -30,7 +30,6 @@ import (
 	plug "github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk"
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/collections"
 	"github.com/kgateway-dev/kgateway/v2/pkg/reports"
-	"github.com/kgateway-dev/kgateway/v2/pkg/syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/utils/stopwatch"
 )
 
@@ -61,9 +60,9 @@ func NewStatusSyncer(
 	reportQueue utils.AsyncQueue[reports.ReportMap],
 	backendPolicyReportQueue utils.AsyncQueue[reports.ReportMap],
 	cacheSyncs []cache.InformerSynced,
-	opts ...syncer.StatusSyncerOption,
+	opts ...StatusSyncerOption,
 ) *StatusSyncer {
-	cfg := syncer.ProcessStatusSyncerOptions(opts...)
+	cfg := processStatusSyncerOptions(opts...)
 	return &StatusSyncer{
 		mgr:                            mgr,
 		plugins:                        plugins,

pkg/kgateway/setup/setup.go

@@ -32,6 +32,7 @@ import (
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/admin"
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/agentgatewaysyncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/controller"
+	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/proxy_syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/wellknown"
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/xds"
 	"github.com/kgateway-dev/kgateway/v2/pkg/krtcollections"
@@ -41,7 +42,6 @@ import (
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/collections"
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/krtutil"
 	"github.com/kgateway-dev/kgateway/v2/pkg/schemes"
-	"github.com/kgateway-dev/kgateway/v2/pkg/syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/utils/envutils"
 	"github.com/kgateway-dev/kgateway/v2/pkg/utils/namespaces"
 	"github.com/kgateway-dev/kgateway/v2/pkg/validator"
@@ -204,12 +204,18 @@ func WithCommonCollectionsOptions(commonCollectionsOptions []collections.Option)
 	}
 }
 
-func WithStatusSyncerOptions(statusSyncerOptions []syncer.StatusSyncerOption) func(*setup) {
+func WithStatusSyncerOptions(statusSyncerOptions []proxy_syncer.StatusSyncerOption) func(*setup) {
 	return func(s *setup) {
 		s.statusSyncerOptions = statusSyncerOptions
 	}
 }
 
+func WithAgentgatewaySyncerOptions(agentgatewaySyncerOptions []agentgatewaysyncer.AgentgatewaySyncerOption) func(*setup) {
+	return func(s *setup) {
+		s.agentgatewaySyncerOptions = agentgatewaySyncerOptions
+	}
+}
+
 type setup struct {
 	apiClient                      apiclient.Client
 	extraInformerCacheSyncHandlers []cache.InformerSynced
@@ -238,8 +244,9 @@ type setup struct {
 	validator                    validator.Validator
 	extraAgwPolicyStatusHandlers map[schema.GroupVersionKind]agwplugins.AgwPolicyStatusSyncHandler
 
-	commonCollectionsOptions []collections.Option
-	statusSyncerOptions      []syncer.StatusSyncerOption
+	commonCollectionsOptions  []collections.Option
+	statusSyncerOptions       []proxy_syncer.StatusSyncerOption
+	agentgatewaySyncerOptions []agentgatewaysyncer.AgentgatewaySyncerOption
 }
 
 var _ Server = &setup{}
@@ -508,6 +515,7 @@ func (s *setup) buildKgatewayWithConfig(
 		ExtraAgwPolicyStatusHandlers: s.extraAgwPolicyStatusHandlers,
 		GatewayControllerExtension:   s.gatewayControllerExtension,
 		StatusSyncerOptions:          s.statusSyncerOptions,
+		AgentgatewaySyncerOptions:    s.agentgatewaySyncerOptions,
 	})
 	if err != nil {
 		slog.Error("failed initializing controller: ", "error", err)

pkg/setup/setup.go

@@ -14,10 +14,11 @@ import (
 	agwplugins "github.com/kgateway-dev/kgateway/v2/pkg/agentgateway/plugins"
 	"github.com/kgateway-dev/kgateway/v2/pkg/apiclient"
 	"github.com/kgateway-dev/kgateway/v2/pkg/deployer"
+	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/agentgatewaysyncer"
+	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/proxy_syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/kgateway/setup"
 	sdk "github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk"
 	"github.com/kgateway-dev/kgateway/v2/pkg/pluginsdk/collections"
-	"github.com/kgateway-dev/kgateway/v2/pkg/syncer"
 	"github.com/kgateway-dev/kgateway/v2/pkg/validator"
 )
 
@@ -50,8 +51,9 @@ type Options struct {
 	// ExtraAgwPolicyStatusHandlers maps policy kinds to their status sync handlers for AgentGateway
 	ExtraAgwPolicyStatusHandlers map[schema.GroupVersionKind]agwplugins.AgwPolicyStatusSyncHandler
 
-	CommonCollectionsOptions []collections.Option
-	StatusSyncerOptions      []syncer.StatusSyncerOption
+	CommonCollectionsOptions  []collections.Option
+	StatusSyncerOptions       []proxy_syncer.StatusSyncerOption
+	AgentGatewaySyncerOptions []agentgatewaysyncer.AgentgatewaySyncerOption
 }
 
 func New(opts Options) (setup.Server, error) {
@@ -78,5 +80,6 @@ func New(opts Options) (setup.Server, error) {
 		setup.WithExtraAgwPolicyStatusHandlers(opts.ExtraAgwPolicyStatusHandlers),
 		setup.WithCommonCollectionsOptions(opts.CommonCollectionsOptions),
 		setup.WithStatusSyncerOptions(opts.StatusSyncerOptions),
+		setup.WithAgentgatewaySyncerOptions(opts.AgentGatewaySyncerOptions),
 	)
 }