lb: serve Gateways (Gateway API v1) from the node holding their address

Gateways of a GatewayClass whose controllerName is reweave.levk.dev/gateway
are served like LoadBalancer Services: a Gateway names one IPv4 address in
spec.addresses, and the node holding that address serves all of it and
writes its status.

- TCP listeners forward the oldest TCPRoute attached to them in the kernel,
  through the same DNAT and reply steering as LoadBalancer Services, so the
  backend sees the client's own address.
- HTTP and HTTPS listeners are served by the agent itself on the holder:
  TLS terminated with the listener's certificate (chosen by SNI, swapped in
  place when its Secret changes, HTTP/2 offered), requests routed by the
  attached HTTPRoutes in the order the spec prescribes (hostname, exact
  path, longest prefix, method, headers, query, age, name, rule), and passed
  to a ready endpoint of the backend with X-Forwarded-For built from this
  hop only. RequestRedirect, URLRewrite and both header modifiers are
  implemented; other filters and regular expressions are refused
  (Accepted False, UnsupportedValue). Bodies are streamed, never limited.
  Unresolved backends answer 500, backends without ready endpoints 503,
  requests nothing matches 404. Those ports are let in ahead of the host
  firewall (REWEAVE-LB-IN).

Listeners are checked for protocol, route kinds, certificates (Secrets in
the Gateway's namespace or granted by a ReferenceGrant, and only in the
namespaces the agent may read: --gateway-namespaces, default
gateway-system), conflicts on a port, and ports LoadBalancer Services
already forward. Routes attach by parentRef (sectionName, port),
allowedRoutes (namespaces Same, All or Selector; kinds) and hostname
intersection, with the spec's conditions and reasons.

One writer per status: the holder writes its Gateway's (the first node by
name when nobody holds the address, so failures are reported once), the
first node by name writes GatewayClass and route statuses, keeping other
controllers' route parents. Writes are conditional on the resourceVersion
they were computed from, and a condition keeps its transition time while
its status stands, so an unchanged status is never rewritten.

The types are plain structs for the subset implemented (the upstream module
needs a newer client-go than the fork's); the CRDs are installed apart from
this manifest, which grants the agent read access to the Gateway API kinds,
patch on their status, and Secrets in gateway-system only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ

lb/controller.go

@@ -53,6 +53,15 @@ type Controller struct {
 	// lastEgress is what was last applied, reused while the EgressService caches are not synced (never wipe
 	// members' rules on a partial view)
 	lastEgress EgressDesired
+
+	// Gateways (nil gateway client: not served). Secrets are read only in gwNamespaces.
+	gateway                                                      rest.Interface
+	gwNamespaces                                                 []string
+	proxy                                                        *Proxy
+	classes, gateways, httpRoutes, tcpRoutes, grants, namespaces cache.Store
+	secrets                                                      []cache.Store
+	gatewaySynced                                                []cache.InformerSynced // empty while the Gateway API is absent
+	unbound                                                      map[string]string      // listeners the proxy could not bind, last pass
 }
 
 // Resync is how often the node re-reads its interfaces and re-applies everything even without API events.
@@ -93,6 +102,86 @@ func (c *Controller) egressAPI() (bool, error) {
 	return false, err
 }
 
+// WithGateway has the controller serve Gateways of the classes whose controllerName is GatewayControllerName,
+// reading their certificates from Secrets in these namespaces only.
+func (c *Controller) WithGateway(client rest.Interface, namespaces []string) *Controller {
+	c.gateway, c.gwNamespaces, c.proxy, c.unbound = client, namespaces, NewProxy(), map[string]string{}
+	return c
+}
+
+// gatewayAPI is which Gateway API resources are served (nil: none, the CRDs are not installed).
+func (c *Controller) gatewayAPI() (map[string]bool, error) {
+	l, err := c.client.Discovery().ServerResourcesForGroupVersion(GatewayGroupVersion.String())
+	if apierrors.IsNotFound(err) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	served := map[string]bool{}
+	for _, r := range l.APIResources {
+		served[r.Name] = true
+	}
+	return served, nil
+}
+
+func (c *Controller) startGateway(ctx context.Context, served map[string]bool) {
+	run := func(store *cache.Store, resource string, obj runtime.Object) {
+		if !served[resource] {
+			return // an older bundle without this kind: nothing of it to serve
+		}
+		var ctl cache.Controller
+		*store, ctl = newInformer(c, cache.NewListWatchFromClient(c.gateway, resource, "", fields.Everything()), obj)
+		go ctl.Run(ctx.Done())
+		c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced)
+	}
+	run(&c.classes, GatewayClasses, &GatewayClass{})
+	run(&c.gateways, Gateways, &Gateway{})
+	run(&c.httpRoutes, HTTPRoutes, &HTTPRoute{})
+	run(&c.tcpRoutes, TCPRoutes, &TCPRoute{})
+	run(&c.grants, ReferenceGrants, &ReferenceGrant{})
+	var ctl cache.Controller
+	c.namespaces, ctl = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "namespaces", "", fields.Everything()), &coreapi.Namespace{})
+	go ctl.Run(ctx.Done())
+	c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced)
+	for _, ns := range c.gwNamespaces {
+		var store cache.Store
+		store, ctl = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "secrets", ns, fields.Everything()), &coreapi.Secret{})
+		go ctl.Run(ctx.Done())
+		c.secrets = append(c.secrets, store)
+		c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced)
+	}
+	common.Log.Infof("lb: serving Gateways of controller %s (certificates from %v)", GatewayControllerName, c.gwNamespaces)
+}
+
+// gatewayReady: the Gateway view can be trusted (the API is served and every informer has synced).
+func (c *Controller) gatewayReady() bool {
+	if c.gateway == nil || len(c.gatewaySynced) == 0 {
+		return false
+	}
+	for _, s := range c.gatewaySynced {
+		if !s() {
+			return false
+		}
+	}
+	return true
+}
+
+// tryStartGateway starts the Gateway informers once the API is served.
+func (c *Controller) tryStartGateway(ctx context.Context) {
+	if c.gateway == nil || len(c.gatewaySynced) != 0 {
+		return
+	}
+	served, err := c.gatewayAPI()
+	if err != nil {
+		common.Log.Warnf("lb: cannot tell whether the Gateway API is served: %v", err)
+		return
+	}
+	if served[Gateways] {
+		c.startGateway(ctx, served)
+	}
+}
+
 func (c *Controller) startEgress(ctx context.Context) {
 	var ctl cache.Controller
 	c.egresses, ctl = newInformer(c, cache.NewListWatchFromClient(c.egress, EgressServices, "", fields.Everything()), &EgressService{})
@@ -130,6 +219,10 @@ func (c *Controller) Run(ctx context.Context) {
 		case <-time.After(5 * time.Second):
 		}
 	}
+	c.tryStartGateway(ctx)
+	if c.proxy != nil {
+		defer c.proxy.Close()
+	}
 	go c.serveAdmission(ctx)
 	if !cache.WaitForCacheSync(ctx.Done(), c.synced...) {
 		return
@@ -143,6 +236,9 @@ func (c *Controller) Run(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-tick.C:
+			c.mu.Lock()
+			c.tryStartGateway(ctx)
+			c.mu.Unlock()
 			if c.egressSynced == nil {
 				// the API may have been installed since
 				if present, err := c.egressAPI(); err == nil && present {
@@ -208,6 +304,14 @@ func (c *Controller) sync(ctx context.Context) error {
 		slices = append(slices, o.(*discovery.EndpointSlice))
 	}
 	d := Compute(c.self, nodes, svcs, slices)
+	var g *GatewayDesired
+	if c.gatewayReady() {
+		gd := ComputeGateways(c.self, nodes, c.gatewayInput(svcs, slices))
+		d = MergeGateways(d, gd, c.self, nodes)
+		g = &gd
+	} else if c.gateway != nil && len(c.gatewaySynced) != 0 {
+		problems = append(problems, "Gateway view not synced yet: Gateways not served")
+	}
 	// rewrite once after start what an earlier run may have written without ipMode
 	for _, f := range d.Forwards {
 		if s := c.service(f.Namespace, f.Name); s != nil && c.written[s.UID] != f.Address.String() && !hasStatusFor(d, f) {
@@ -236,6 +340,18 @@ func (c *Controller) sync(ctx context.Context) error {
 		return fmt.Errorf("applying datapath: %v", err)
 	}
 	c.lastEgress = e
+	if g != nil {
+		// the ports are let in by now; bind what this node serves, close what it no longer does
+		unbound := map[string]string{}
+		for k, err := range c.proxy.Apply(g.Serve) {
+			unbound[k] = err.Error()
+		}
+		if !sameJSON(unbound, c.unbound) {
+			c.unbound = unbound
+			c.poke() // the statuses say so on the next pass
+		}
+		c.writeGatewayStatuses(ctx, *g)
+	}
 	for _, st := range d.Status {
 		if err := c.writeStatus(ctx, st); err != nil {
 			return err
@@ -458,3 +574,56 @@ func (c *Controller) writeStatus(ctx context.Context, st Status) error {
 	}
 	return nil
 }
+
+// gatewayInput is the Gateway view the policy reads.
+func (c *Controller) gatewayInput(svcs []*coreapi.Service, slices []*discovery.EndpointSlice) GatewayInput {
+	in := GatewayInput{Services: svcs, Slices: slices, SecretNamespaces: map[string]bool{}, Unbound: c.unbound, Now: metav1.Now()}
+	list := func(s cache.Store, each func(interface{})) {
+		if s != nil {
+			for _, o := range s.List() {
+				each(o)
+			}
+		}
+	}
+	list(c.classes, func(o interface{}) { in.Classes = append(in.Classes, o.(*GatewayClass)) })
+	list(c.gateways, func(o interface{}) { in.Gateways = append(in.Gateways, o.(*Gateway)) })
+	list(c.httpRoutes, func(o interface{}) { in.HTTPRoutes = append(in.HTTPRoutes, o.(*HTTPRoute)) })
+	list(c.tcpRoutes, func(o interface{}) { in.TCPRoutes = append(in.TCPRoutes, o.(*TCPRoute)) })
+	list(c.grants, func(o interface{}) { in.Grants = append(in.Grants, o.(*ReferenceGrant)) })
+	list(c.namespaces, func(o interface{}) { in.Namespaces = append(in.Namespaces, o.(*coreapi.Namespace)) })
+	for _, s := range c.secrets {
+		list(s, func(o interface{}) { in.Secrets = append(in.Secrets, o.(*coreapi.Secret)) })
+	}
+	for _, ns := range c.gwNamespaces {
+		in.SecretNamespaces[ns] = true
+	}
+	return in
+}
+
+// writeGatewayStatuses writes what the policy says this node writes. Each write is conditional on the version it
+// was computed from: one built from a stale view fails, and the next pass recomputes it.
+func (c *Controller) writeGatewayStatuses(ctx context.Context, g GatewayDesired) {
+	for _, w := range g.Classes {
+		if err := patchStatus(ctx, c.gateway, GatewayClasses, "", w.Name, w.ResourceVersion, w.Status); err != nil {
+			common.Log.Warnf("lb: status of GatewayClass %s: %v", w.Name, err)
+			continue
+		}
+		common.Log.Infof("lb: GatewayClass %s accepted", w.Name)
+	}
+	for _, w := range g.Gateways {
+		if err := patchStatus(ctx, c.gateway, Gateways, w.Namespace, w.Name, w.ResourceVersion, w.Status); err != nil {
+			common.Log.Warnf("lb: status of Gateway %s/%s: %v", w.Namespace, w.Name, err)
+			continue
+		}
+		var conds []string
+		for _, cd := range w.Status.Conditions {
+			conds = append(conds, fmt.Sprintf("%s=%s (%s)", cd.Type, cd.Status, cd.Reason))
+		}
+		common.Log.Infof("lb: Gateway %s/%s: %v", w.Namespace, w.Name, conds)
+	}
+	for _, w := range g.Routes {
+		if err := patchStatus(ctx, c.gateway, w.Resource, w.Namespace, w.Name, w.ResourceVersion, w.Status); err != nil {
+			common.Log.Warnf("lb: status of %s %s/%s: %v", w.Resource, w.Namespace, w.Name, err)
+		}
+	}
+}