Rework fast-path provisioning to per-Gateway data-plane Deployment

Address review feedback on gke-labs#624:
- Provision a dedicated data-plane Deployment per Gateway in pkg/provisioning/singlepod selecting its own LoadBalancer Service.
- Support data-plane mode in GARI with GatewayFilter and DisableStatusUpdates.
- Derive Service ports from Gateway effective listeners (including ListenerSets).
- Gate Programmed condition on Deployment availability and Service LoadBalancer address assignment without fallback.
- Implement label-based orphan sweeping and deletion cleanup via GatewayDeleteHandler.OnGatewayDeleted.
- Drop dynamic listener manager, port allocator, scoped proxy, and Gateway finalizer.

Fixes gke-labs#620

Author: codebot-robot

.agents/skills/implement-conformance-test/journal/httproute-multiple-gateways.md

@@ -7,25 +7,27 @@
 
 ## 2. Issue / Failure Analysis
 - **Observed Behavior**: In single-pod fast-path mode, all Gateways previously shared a single `default/gari-proxy` LoadBalancer Service and shared proxy ports (`:8000`, `:8443`), routing solely by Host header. Gateways could not be isolated when two Gateways had overlapping catch-all paths (e.g., both having `/` on port 80).
-- **Root Cause**: Fast-path mode lacked per-Gateway LoadBalancer Service provisioning and distinct pod port allocation for listener isolation.
+- **Root Cause**: Fast-path mode lacked per-Gateway data-plane Deployment provisioning and per-Gateway LoadBalancer Services for traffic and listener isolation.
 
 ## 3. Implementation / Fix Strategy
 - **Approach**:
-  1. Implemented per-Gateway `type: LoadBalancer` Service provisioning in `pkg/provisioning/singlepod` with deterministic naming (`gari-gw-<hash>`), Gateway identifying labels, and pod port allocation in range `20000-29999`.
-  2. Maintained stable port allocations across restarts seeded from existing Services, and implemented clean finalizer/GC cleanup upon Gateway deletion.
-  3. Added `ListenerAddressProvider` dynamic listener management in `pkg/gari` and scoped proxy routing in `pkg/proxy` so connections arriving on a dynamic listener port are evaluated only against that Gateway's listeners, certificates, and routes.
-  4. Preserved shared-listener fallback when no `ListenerAddressProvider` is configured (for embedders like `snigateway`).
-  5. Removed the shared `gari-proxy` Service from `k8s/controller.yaml`, updated RBAC for Service CRUD, and updated e2e harness and tests to use Gateway `status.addresses`.
-  6. Enabled `tests.HTTPRouteMultipleGateways` in `tests/e2e/conformance_test.go`.
+  1. Implemented per-Gateway Deployment and `type: LoadBalancer` Service provisioning in `pkg/provisioning/singlepod` with deterministic naming (`gari-gw-<hash>`), Gateway identifying labels (`gateway.networking.k8s.io/gateway-name`, `gateway.networking.k8s.io/gateway-namespace`, `app.kubernetes.io/managed-by: gari-singlepod`), and service selector matching its dedicated Deployment.
+  2. Derived Service ports dynamically from effective listeners (including ListenerSets).
+  3. Added data-plane mode (`--dataplane-mode`, `--gateway-namespace`, `--gateway-name`, `DisableStatusUpdates: true`) so each per-Gateway pod reconciles and serves only its designated Gateway without colliding on Host headers or writing status back.
+  4. Gated Gateway `Programmed` condition on the provisioned Deployment being ready/available and the Service having an assigned LoadBalancer address (with no fallback).
+  5. Implemented label-based orphan sweeping and deletion cleanup via `GatewayDeleteHandler.OnGatewayDeleted` instead of cross-namespace finalizers.
+  6. Removed the shared `gari-proxy` Service from `k8s/controller.yaml`, granted Deployment and Service CRUD permissions in ClusterRole RBAC, and updated e2e harness and tests to use Gateway `status.addresses`.
+  7. Enabled `tests.HTTPRouteMultipleGateways` in `tests/e2e/conformance_test.go`.
 - **Key Files Modified**:
   - `pkg/provisioning/singlepod/address_provider.go`
   - `pkg/provisioning/singlepod/address_provider_test.go`
   - `pkg/controller/gateway_controller.go`
+  - `pkg/controller/httproute_controller.go`
+  - `pkg/controller/listenerset_controller.go`
+  - `pkg/controller/backendtlspolicy_controller.go`
   - `pkg/controller/utils.go`
-  - `pkg/controller/httproute_controller.go` (and other reconcilers)
-  - `pkg/proxy/proxy.go`
-  - `pkg/proxy/proxy_test.go`
-  - `pkg/gari/dynamic_listeners.go`
+  - `pkg/state/compiled.go`
+  - `pkg/state/compiled_test.go`
   - `pkg/gari/gari.go`
   - `cmd/gateway-api-reference-implementation/main.go`
   - `k8s/controller.yaml`
@@ -34,7 +36,7 @@
   - `tests/e2e/conformance_test.go`
 
 ## 4. Validation & Results
-- **Unit Tests**: Added port allocator tests, singlepod address provider tests, and scoped proxy routing tests; verified with `ap test`.
+- **Unit Tests**: Added unit tests for per-Gateway Deployment & Service provisioning, effective listener ports, readiness gating, and orphan sweeping; verified with `ap test` and `ap lint`.
 - **Conformance Logs**:
   ```
   --- PASS: TestConformance (102.75s)

cmd/gateway-api-reference-implementation/main.go

@@ -22,20 +22,37 @@ import (
 	"github.com/gke-labs/gateway-api-reference-implementation/pkg/provisioning/singlepod"
 	"k8s.io/klog/v2/textlogger"
 	ctrl "sigs.k8s.io/controller-runtime"
+	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 )
 
 var setupLog = ctrl.Log.WithName("setup")
 
 func main() {
 	opts := gari.DefaultOptions()
 
-	var controllerNamespace string
+	var (
+		controllerNamespace string
+		dataplaneMode       bool
+		gatewayNamespace    string
+		gatewayName         string
+		dataplaneImage      string
+	)
+
 	defaultNs := os.Getenv("POD_NAMESPACE")
 	if defaultNs == "" {
 		defaultNs = singlepod.DefaultControllerNamespace
 	}
 
-	flag.StringVar(&controllerNamespace, "controller-namespace", defaultNs, "The namespace of the controller where per-Gateway Services are provisioned.")
+	defaultImg := os.Getenv("GARI_IMAGE")
+	if defaultImg == "" {
+		defaultImg = singlepod.DefaultDataplaneImage
+	}
+
+	flag.StringVar(&controllerNamespace, "controller-namespace", defaultNs, "The namespace of the controller where per-Gateway Services and Deployments are provisioned.")
+	flag.BoolVar(&dataplaneMode, "dataplane-mode", false, "Run in data-plane mode serving a single Gateway.")
+	flag.StringVar(&gatewayNamespace, "gateway-namespace", "", "The namespace of the Gateway to serve in data-plane mode.")
+	flag.StringVar(&gatewayName, "gateway-name", "", "The name of the Gateway to serve in data-plane mode.")
+	flag.StringVar(&dataplaneImage, "dataplane-image", defaultImg, "The container image to use for provisioned per-Gateway data-plane Deployments.")
 	flag.StringVar(&opts.MetricsAddr, "metrics-bind-address", opts.MetricsAddr, "The address the metric endpoint binds to.")
 	flag.StringVar(&opts.HealthProbeBindAddress, "health-probe-bind-address", opts.HealthProbeBindAddress, "The address the probe endpoint binds to.")
 	flag.StringVar(&opts.ProxyAddr, "proxy-bind-address", opts.ProxyAddr, "The address the proxy binds to.")
@@ -54,7 +71,27 @@ func main() {
 
 	ctrl.SetLogger(textlogger.NewLogger(logConfig))
 
-	opts.AddressProvider = singlepod.NewAddressProvider(nil, controllerNamespace)
+	if dataplaneMode {
+		opts.DisableStatusUpdates = true
+		opts.LeaderElection = false
+		opts.ProxyHTTP3Addr = opts.ProxyHTTPSAddr // enable HTTP/3 on the HTTPS port for dataplane
+		opts.AddressProvider = nil
+		if gatewayNamespace != "" && gatewayName != "" {
+			opts.GatewayFilter = func(gw *gatewayv1.Gateway) bool {
+				return gw.Namespace == gatewayNamespace && gw.Name == gatewayName
+			}
+		}
+	} else {
+		opts.ProxyAddr = ""
+		opts.ProxyHTTPSAddr = ""
+		opts.ProxyHTTP3Addr = ""
+		opts.AddressProvider = singlepod.NewAddressProvider(
+			nil,
+			controllerNamespace,
+			singlepod.WithDataplaneImage(dataplaneImage),
+			singlepod.WithEnableH2C(opts.EnableH2C),
+		)
+	}
 
 	ctx := ctrl.SetupSignalHandler()
 	if err := gari.Run(ctx, opts); err != nil {

k8s/controller.yaml

@@ -15,6 +15,9 @@ rules:
 - apiGroups: [""]
   resources: ["services"]
   verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
+- apiGroups: ["apps"]
+  resources: ["deployments"]
+  verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
 - apiGroups: [""]
   resources: ["configmaps", "secrets", "namespaces"]
   verbs: ["get", "list", "watch"]

pkg/controller/backendtlspolicy_controller.go

@@ -37,12 +37,12 @@ import (
 
 type BackendTLSPolicyReconciler struct {
 	client.Client
-	Scheme           *runtime.Scheme
-	State            *state.State
-	Proxy            *proxy.Proxy
-	ControllerName   string
-	OnGatewaysUpdate func([]*gatewayv1.Gateway)
-	ModelUpdater     ModelUpdater
+	Scheme               *runtime.Scheme
+	State                *state.State
+	Proxy                *proxy.Proxy
+	ControllerName       string
+	OnGatewaysUpdate     func([]*gatewayv1.Gateway)
+	DisableStatusUpdates bool
 }
 
 func (r *BackendTLSPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -267,7 +267,7 @@ func (r *BackendTLSPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		}
 	}
 
-	if updated {
+	if updated && !r.DisableStatusUpdates {
 		policy.Status.Ancestors = ancestors
 		if err := r.Status().Update(ctx, policy); err != nil {
 			l.Error(err, "unable to update BackendTLSPolicy status")
@@ -285,7 +285,7 @@ func (r *BackendTLSPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req
 }
 
 func (r *BackendTLSPolicyReconciler) updateProxy() {
-	updateProxy(r.State, r.Proxy, r.ControllerName, r.OnGatewaysUpdate, r.ModelUpdater)
+	updateProxy(r.State, r.Proxy, r.ControllerName, r.OnGatewaysUpdate)
 }
 
 func (r *BackendTLSPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {

pkg/controller/configmap_controller.go

@@ -38,7 +38,6 @@ type ConfigMapReconciler struct {
 	Proxy            *proxy.Proxy
 	ControllerName   string
 	OnGatewaysUpdate func([]*gatewayv1.Gateway)
-	ModelUpdater     ModelUpdater
 }
 
 func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -62,7 +61,7 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 }
 
 func (r *ConfigMapReconciler) updateProxy() {
-	updateProxy(r.State, r.Proxy, r.ControllerName, r.OnGatewaysUpdate, r.ModelUpdater)
+	updateProxy(r.State, r.Proxy, r.ControllerName, r.OnGatewaysUpdate)
 }
 
 func (r *ConfigMapReconciler) SetupWithManager(mgr ctrl.Manager) error {

pkg/controller/controller_test.go

@@ -910,11 +910,16 @@ func findCondition(conditions []metav1.Condition, condType string) *metav1.Condi
 
 type fakeAddressProvider struct {
 	addresses []gatewayv1.GatewayStatusAddress
+	ready     bool
 	err       error
 }
 
-func (f *fakeAddressProvider) GatewayAddresses(ctx context.Context, gw *gatewayv1.Gateway) ([]gatewayv1.GatewayStatusAddress, error) {
-	return f.addresses, f.err
+func (f *fakeAddressProvider) GatewayAddresses(ctx context.Context, gw *gatewayv1.Gateway, effectiveListeners []*state.EffectiveListener) ([]gatewayv1.GatewayStatusAddress, bool, error) {
+	ready := f.ready
+	if !ready && f.err == nil && len(f.addresses) > 0 {
+		ready = true
+	}
+	return f.addresses, ready, f.err
 }
 
 type fakeAddressWatcherProvider struct {