diff --git a/lb/controller.go b/lb/controller.go index bfcdf05f..2db86057 100644 --- a/lb/controller.go +++ b/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) + } + } +} diff --git a/lb/datapath.go b/lb/datapath.go index 91e19541..5973a36e 100644 --- a/lb/datapath.go +++ b/lb/datapath.go @@ -117,6 +117,10 @@ func Render(d Desired, e EgressDesired, self string, podNet *net.IPNet, stale [] for _, c := range []string{FwdChain, InChain} { fmt.Fprintf(&filter, ":%s - [0:0]\n", c) } + for _, a := range d.Accept { + // served by the agent itself (a Gateway's HTTP or HTTPS listener): in, whatever the host firewall says + fmt.Fprintf(&filter, "-A %s -d %s/32 -p tcp -m tcp --dport %d -m comment --comment %q -j ACCEPT\n", InChain, a.Address, a.Port, a.Comment) + } renderEgress(e, self, podNet, &body, &mangle, &filter) nat.Write(body.Bytes()) diff --git a/lb/gateway.go b/lb/gateway.go new file mode 100644 index 00000000..b047af5b --- /dev/null +++ b/lb/gateway.go @@ -0,0 +1,1009 @@ +package lb + +import ( + "crypto/tls" + "encoding/json" + "fmt" + "net" + "sort" + "strings" + "time" + + coreapi "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" +) + +// Gateways of a class this agent implements are served like LoadBalancer Services: a Gateway names one address a +// node holds (spec.addresses), and that node, the holder, serves all of it and writes its status. TCP listeners +// are forwarded in the kernel (the DNAT of LoadBalancer Services: 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, requests routed by their HTTPRoutes). ComputeGateways is the whole policy as a pure function of +// what the API shows. + +const ( + ProtocolHTTP = "HTTP" + ProtocolHTTPS = "HTTPS" + ProtocolTCP = "TCP" + + kindGateway = "Gateway" + kindHTTPRoute = "HTTPRoute" + kindTCPRoute = "TCPRoute" + kindService = "Service" + kindSecret = "Secret" + + condAccepted = "Accepted" + condProgrammed = "Programmed" + condResolvedRefs = "ResolvedRefs" + condConflicted = "Conflicted" + + filterRequestHeaderModifier = "RequestHeaderModifier" + filterResponseHeaderModifier = "ResponseHeaderModifier" + filterRequestRedirect = "RequestRedirect" + filterURLRewrite = "URLRewrite" + + pathExact = "Exact" + pathPrefix = "PathPrefix" +) + +// GatewayInput is what the API shows that the gateway policy reads. +type GatewayInput struct { + Classes []*GatewayClass + Gateways []*Gateway + HTTPRoutes []*HTTPRoute + TCPRoutes []*TCPRoute + Grants []*ReferenceGrant + Namespaces []*coreapi.Namespace + Services []*coreapi.Service + Slices []*discovery.EndpointSlice + // Secrets the agent can read, and the namespaces they come from: a certificate elsewhere cannot be served + Secrets []*coreapi.Secret + SecretNamespaces map[string]bool + // Unbound are the listeners ("namespace/gateway/listener") this node could not bind on its last try, and why + Unbound map[string]string + Now metav1.Time +} + +// GatewayDesired is what one node does for Gateways, and the statuses it writes. +type GatewayDesired struct { + Forwards []Forward // this node's TCP listeners, forwarded like LoadBalancer Services + Holding []string // nodes forwarding some Gateway's TCP listener (other nodes steer replies back to them) + Serve []ServeListener // this node's HTTP and HTTPS listeners + Classes []ClassStatusWrite + Gateways []GatewayStatusWrite + Routes []RouteStatusWrite + Problems []string +} + +// ServeListener is an HTTP or HTTPS listener this node serves. +type ServeListener struct { + Gateway string // namespace/name + Name string + Address net.IP + Port int32 + HTTPS bool + Hostname string // "" takes any host + Cert *coreapi.Secret // HTTPS: the listener's certificate + Routes []ServeRoute +} + +// ServeRoute is an HTTPRoute attached to a listener. +type ServeRoute struct { + Key string // namespace/name + Created time.Time + // Hostnames the route answers on this listener (its own, narrowed to the listener's); none: the listener's + Hostnames []string + Rules []ServeRule +} + +// ServeRule is one rule of an attached route. +type ServeRule struct { + Matches []HTTPRouteMatch // defaulted: a match without a path is a PathPrefix match on "/" + Filters []HTTPRouteFilter + Backends []ServeBackend + // Broken: some backend could not be resolved; requests the rule takes are answered 500, as the spec says + Broken bool +} + +// ServeBackend is a Service port a rule sends requests to, with its ready endpoints. +type ServeBackend struct { + Key string // namespace/name:port + Weight int32 + Endpoints []Endpoint + Filters []HTTPRouteFilter +} + +type ClassStatusWrite struct { + Name, ResourceVersion string + Status GatewayClassStatus +} + +type GatewayStatusWrite struct { + Namespace, Name, ResourceVersion string + Status GatewayStatus +} + +type RouteStatusWrite struct { + Resource, Namespace, Name, ResourceVersion string + Status RouteStatus +} + +// listener is a listener as the policy judges it. +type listener struct { + spec Listener + kinds []string // route kinds it takes + usable bool // accepted, references resolved, port free: it can be served unless it conflicts + unbound bool // served, but this node could not bind it on its last try + conds []metav1.Condition + conflicted bool + cert *coreapi.Secret + attached int32 + tcpRoute *TCPRoute // the TCPRoute a TCP listener forwards (the oldest of those attached) + http []ServeRoute +} + +type gatewayView struct { + gw *Gateway + address net.IP + addrErr string + holder *Node + holderErr string + listeners []*listener +} + +// ComputeGateways decides what node self does for Gateways, given every node and what the API shows. taken are +// the address:protocol:port triples LoadBalancer Services already forward (a listener cannot share one). +func ComputeGateways(self string, nodes []Node, in GatewayInput) GatewayDesired { + var d GatewayDesired + sorted := append([]Node(nil), nodes...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) + first := "" + if len(sorted) > 0 { + first = sorted[0].Name + } + holdersOf := func(ip net.IP) []Node { + var hs []Node + for _, n := range sorted { + if n.Holds(ip) { + hs = append(hs, n) + } + } + return hs + } + + ours := map[string]*GatewayClass{} + classes := append([]*GatewayClass(nil), in.Classes...) + sort.Slice(classes, func(i, j int) bool { return classes[i].Name < classes[j].Name }) + for _, c := range classes { + if c.Spec.ControllerName != GatewayControllerName { + continue + } + ours[c.Name] = c + want := GatewayClassStatus{Conditions: []metav1.Condition{ + condition(c.Status.Conditions, condAccepted, true, "Accepted", "served by the reweave agent on the node holding each Gateway's address", c.Generation, in.Now), + }} + if self == first && !sameJSON(want, c.Status) { + d.Classes = append(d.Classes, ClassStatusWrite{Name: c.Name, ResourceVersion: c.ResourceVersion, Status: want}) + } + } + + nsLabels := map[string]labels.Set{} + for _, n := range in.Namespaces { + nsLabels[n.Name] = labels.Set(n.Labels) + } + secrets := map[string]*coreapi.Secret{} + for _, s := range in.Secrets { + secrets[s.Namespace+"/"+s.Name] = s + } + services := map[string]*coreapi.Service{} + taken := map[string]string{} + for _, s := range in.Services { + services[s.Namespace+"/"+s.Name] = s + if !eligible(s) { + continue + } + if ip := net.ParseIP(s.Spec.LoadBalancerIP).To4(); ip != nil { + for _, p := range s.Spec.Ports { + proto := p.Protocol + if proto == "" { + proto = coreapi.ProtocolTCP + } + taken[fmt.Sprintf("%s:%s:%d", ip, proto, p.Port)] = key(s) + } + } + } + + // the Gateways of our classes, judged listener by listener + var views []*gatewayView + byKey := map[string]*gatewayView{} + gws := append([]*Gateway(nil), in.Gateways...) + sort.Slice(gws, func(i, j int) bool { return objKey(&gws[i].ObjectMeta) < objKey(&gws[j].ObjectMeta) }) + for _, gw := range gws { + if ours[gw.Spec.GatewayClassName] == nil { + continue + } + v := &gatewayView{gw: gw} + v.address, v.addrErr = gatewayAddress(gw) + if v.address != nil { + switch hs := holdersOf(v.address); len(hs) { + case 0: + v.holderErr = fmt.Sprintf("no node holds %s", v.address) + case 1: + v.holder = &hs[0] + default: + names := make([]string, len(hs)) + for i, h := range hs { + names[i] = h.Name + } + v.holderErr = fmt.Sprintf("%s is held by several nodes (%s); serving it from none", v.address, strings.Join(names, ", ")) + } + } + v.listeners = judgeListeners(gw, v.address, in, secrets, taken) + views = append(views, v) + byKey[objKey(&gw.ObjectMeta)] = v + if v.addrErr != "" { + d.Problems = append(d.Problems, fmt.Sprintf("gateway %s: %s", objKey(&gw.ObjectMeta), v.addrErr)) + } else if v.holderErr != "" { + d.Problems = append(d.Problems, fmt.Sprintf("gateway %s: %s", objKey(&gw.ObjectMeta), v.holderErr)) + } + } + + // routes, oldest first: that order decides which TCPRoute a TCP listener forwards, and HTTP precedence ties + type routeRef struct { + resource, kind string + meta *metav1.ObjectMeta + parents []ParentReference + current RouteStatus + http *HTTPRoute + tcp *TCPRoute + } + var routes []routeRef + for _, r := range in.HTTPRoutes { + routes = append(routes, routeRef{resource: HTTPRoutes, kind: kindHTTPRoute, meta: &r.ObjectMeta, parents: r.Spec.ParentRefs, current: r.Status, http: r}) + } + for _, r := range in.TCPRoutes { + routes = append(routes, routeRef{resource: TCPRoutes, kind: kindTCPRoute, meta: &r.ObjectMeta, parents: r.Spec.ParentRefs, current: r.Status, tcp: r}) + } + sort.Slice(routes, func(i, j int) bool { + a, b := routes[i], routes[j] + if !a.meta.CreationTimestamp.Equal(&b.meta.CreationTimestamp) { + return a.meta.CreationTimestamp.Before(&b.meta.CreationTimestamp) + } + if a.kind != b.kind { + return a.kind < b.kind + } + return objKey(a.meta) < objKey(b.meta) + }) + for _, r := range routes { + var ourParents []RouteParentStatus + for _, pr := range r.parents { + if group(pr.Group, GatewayGroup) != GatewayGroup || kind(pr.Kind, kindGateway) != kindGateway { + continue + } + v := byKey[ns(pr.Namespace, r.meta.Namespace)+"/"+pr.Name] + if v == nil { + continue // not one of ours (another implementation's, or none's) + } + var accepted metav1.Condition + var resolved metav1.Condition + if r.http != nil { + accepted, resolved = attachHTTP(r.http, pr, v, in, nsLabels, services) + } else { + accepted, resolved = attachTCP(r.tcp, pr, v, in, nsLabels, services) + } + ourParents = append(ourParents, RouteParentStatus{ + ParentRef: pr, + ControllerName: GatewayControllerName, + Conditions: []metav1.Condition{ + condition(currentParent(r.current, pr), condAccepted, accepted.Status == metav1.ConditionTrue, accepted.Reason, accepted.Message, r.meta.Generation, in.Now), + condition(currentParent(r.current, pr), condResolvedRefs, resolved.Status == metav1.ConditionTrue, resolved.Reason, resolved.Message, r.meta.Generation, in.Now), + }, + }) + } + // our entries are rewritten whole; other controllers' entries are kept as they are + var parents []RouteParentStatus + hadOurs := false + for _, p := range r.current.Parents { + if p.ControllerName == GatewayControllerName { + hadOurs = true + continue + } + parents = append(parents, p) + } + parents = append(parents, ourParents...) + if len(ourParents) == 0 && !hadOurs { + continue + } + want := RouteStatus{Parents: parents} + if want.Parents == nil { + want.Parents = []RouteParentStatus{} + } + if self == first && !sameJSON(want, r.current) { + d.Routes = append(d.Routes, RouteStatusWrite{Resource: r.resource, Namespace: r.meta.Namespace, Name: r.meta.Name, ResourceVersion: r.meta.ResourceVersion, Status: want}) + } + } + + // what this node serves, and every Gateway's status from its writer + holding := map[string]bool{} + for _, v := range views { + gw := v.gw + served := v.holder != nil + if served { + for _, l := range v.listeners { + if l.usable && !l.conflicted && l.spec.Protocol == ProtocolTCP && l.tcpRoute != nil { + holding[v.holder.Name] = true + } + } + } + if served && v.holder.Name == self { + for _, l := range v.listeners { + if !l.usable || l.conflicted { + continue + } + switch l.spec.Protocol { + case ProtocolTCP: + if l.tcpRoute == nil { + continue + } + d.Forwards = append(d.Forwards, Forward{ + Namespace: gw.Namespace, Name: gw.Name + "/" + l.spec.Name, Address: v.address, + Ports: []Port{{Protocol: coreapi.ProtocolTCP, Port: l.spec.Port, Endpoints: tcpEndpoints(l.tcpRoute, in, services)}}, + }) + case ProtocolHTTP, ProtocolHTTPS: + d.Serve = append(d.Serve, ServeListener{ + Gateway: objKey(&gw.ObjectMeta), Name: l.spec.Name, Address: v.address, Port: l.spec.Port, + HTTPS: l.spec.Protocol == ProtocolHTTPS, Hostname: str(l.spec.Hostname), Cert: l.cert, Routes: l.http, + }) + } + } + } + writer := first + if v.holder != nil { + writer = v.holder.Name + } + if writer != self { + continue + } + want := gatewayStatus(v, in.Now) + if !sameJSON(want, gw.Status) { + d.Gateways = append(d.Gateways, GatewayStatusWrite{Namespace: gw.Namespace, Name: gw.Name, ResourceVersion: gw.ResourceVersion, Status: want}) + } + } + for n := range holding { + d.Holding = append(d.Holding, n) + } + sort.Strings(d.Holding) + return d +} + +// gatewayAddress is the one IPv4 address a Gateway names, or why it names none this agent can serve. +func gatewayAddress(gw *Gateway) (net.IP, string) { + if len(gw.Spec.Addresses) != 1 { + return nil, fmt.Sprintf("spec.addresses must name exactly one IPv4 address a node holds (has %d)", len(gw.Spec.Addresses)) + } + a := gw.Spec.Addresses[0] + if t := str(a.Type); t != "" && t != "IPAddress" { + return nil, fmt.Sprintf("address type %s is not supported (IPAddress only)", t) + } + ip := net.ParseIP(a.Value).To4() + if ip == nil { + return nil, fmt.Sprintf("%q is not an IPv4 address", a.Value) + } + return ip, "" +} + +// judgeListeners checks each listener on its own (protocol, route kinds, certificate) and against the others on +// its port (conflicts), and against LoadBalancer Services forwarding the same address and port. +func judgeListeners(gw *Gateway, addr net.IP, in GatewayInput, secrets map[string]*coreapi.Secret, taken map[string]string) []*listener { + gen := gw.Generation + var out []*listener + for _, spec := range gw.Spec.Listeners { + l := &listener{spec: spec, usable: true} + cur := currentListener(gw, spec.Name) + resolved := condition(cur, condResolvedRefs, true, "ResolvedRefs", "", gen, in.Now) + accepted := condition(cur, condAccepted, true, "Accepted", "", gen, in.Now) + var supported []string + switch spec.Protocol { + case ProtocolHTTP, ProtocolHTTPS: + supported = []string{kindHTTPRoute} + case ProtocolTCP: + supported = []string{kindTCPRoute} + default: + l.usable = false + accepted = condition(cur, condAccepted, false, "UnsupportedProtocol", fmt.Sprintf("protocol %s is not supported (HTTP, HTTPS and TCP are)", spec.Protocol), gen, in.Now) + } + l.kinds = supported + if spec.AllowedRoutes != nil && len(spec.AllowedRoutes.Kinds) > 0 { + l.kinds = nil + var bad []string + for _, k := range spec.AllowedRoutes.Kinds { + if group(k.Group, GatewayGroup) == GatewayGroup && contains(supported, k.Kind) { + l.kinds = append(l.kinds, k.Kind) + } else { + bad = append(bad, k.Kind) + } + } + if len(bad) > 0 { + resolved = condition(cur, condResolvedRefs, false, "InvalidRouteKinds", fmt.Sprintf("route kinds %s cannot attach to a %s listener", strings.Join(bad, ", "), spec.Protocol), gen, in.Now) + } + } + if spec.Protocol == ProtocolHTTPS && l.usable { + cert, reason, msg := listenerCert(gw, spec, in, secrets) + if cert == nil { + // without its certificate the listener cannot be served (not programmed); an unsupported TLS mode + // is not accepted at all, an unresolved certificate is a reference problem + l.usable = false + if reason == "UnsupportedValue" { + accepted = condition(cur, condAccepted, false, reason, msg, gen, in.Now) + } else { + resolved = condition(cur, condResolvedRefs, false, reason, msg, gen, in.Now) + } + } + l.cert = cert + } + if addr != nil && l.usable { + if svc, ok := taken[fmt.Sprintf("%s:TCP:%d", addr, spec.Port)]; ok { + l.usable = false + accepted = condition(cur, condAccepted, false, "PortUnavailable", fmt.Sprintf("LoadBalancer Service %s already forwards %s:%d", svc, addr, spec.Port), gen, in.Now) + } else if why, ok := in.Unbound[objKey(&gw.ObjectMeta)+"/"+spec.Name]; ok { + // still served (each pass tries again), but not reported as working + l.unbound = true + accepted = condition(cur, condAccepted, false, "PortUnavailable", "binding "+addr.String()+": "+why, gen, in.Now) + } + } + l.conds = []metav1.Condition{accepted, resolved} + out = append(out, l) + } + // listeners sharing a port must agree on the protocol, and HTTP(S) ones must differ in hostname + byPort := map[int32][]*listener{} + for _, l := range out { + byPort[l.spec.Port] = append(byPort[l.spec.Port], l) + } + for _, ls := range byPort { + protos := map[string]bool{} + for _, l := range ls { + protos[l.spec.Protocol] = true + } + for _, l := range ls { + reason := "" + if len(protos) > 1 { + reason = "ProtocolConflict" + } else { + for _, o := range ls { + if o != l && (l.spec.Protocol == ProtocolTCP || str(o.spec.Hostname) == str(l.spec.Hostname)) { + reason = "HostnameConflict" + } + } + } + if reason != "" { + l.conflicted = true + } + cur := currentListener(gw, l.spec.Name) + if reason != "" { + l.conds = append(l.conds, condition(cur, condConflicted, true, reason, fmt.Sprintf("port %d is shared with listeners that cannot be told apart", l.spec.Port), gen, in.Now)) + } else { + l.conds = append(l.conds, condition(cur, condConflicted, false, "NoConflicts", "", gen, in.Now)) + } + } + } + return out +} + +// listenerCert resolves an HTTPS listener's certificate (the first certificateRef: one certificate per listener). +func listenerCert(gw *Gateway, l Listener, in GatewayInput, secrets map[string]*coreapi.Secret) (*coreapi.Secret, string, string) { + if l.TLS == nil || len(l.TLS.CertificateRefs) == 0 { + return nil, "InvalidCertificateRef", "an HTTPS listener needs tls.certificateRefs" + } + if m := str(l.TLS.Mode); m != "" && m != "Terminate" { + return nil, "UnsupportedValue", fmt.Sprintf("tls mode %s is not supported on HTTPS (Terminate is)", m) + } + ref := l.TLS.CertificateRefs[0] + if group(ref.Group, "") != "" || kind(ref.Kind, kindSecret) != kindSecret { + return nil, "InvalidCertificateRef", fmt.Sprintf("certificateRef %s/%s is not a Secret", group(ref.Group, ""), kind(ref.Kind, kindSecret)) + } + target := ns(ref.Namespace, gw.Namespace) + if target != gw.Namespace && !granted(in.Grants, GatewayGroup, kindGateway, gw.Namespace, "", kindSecret, target, ref.Name) { + return nil, "RefNotPermitted", fmt.Sprintf("no ReferenceGrant in %s lets Gateways in %s use Secret %s", target, gw.Namespace, ref.Name) + } + if !in.SecretNamespaces[target] { + return nil, "RefNotPermitted", fmt.Sprintf("this agent reads Secrets only in its gateway namespaces, not in %s", target) + } + s := secrets[target+"/"+ref.Name] + if s == nil { + return nil, "InvalidCertificateRef", fmt.Sprintf("Secret %s/%s not found", target, ref.Name) + } + if _, err := tls.X509KeyPair(s.Data[coreapi.TLSCertKey], s.Data[coreapi.TLSPrivateKeyKey]); err != nil { + return nil, "InvalidCertificateRef", fmt.Sprintf("Secret %s/%s holds no usable certificate and key: %v", target, ref.Name, err) + } + return s, "", "" +} + +// listenersFor are the listeners a parentRef names (all, or the one its sectionName and port pick). +func listenersFor(v *gatewayView, pr ParentReference) []*listener { + var out []*listener + for _, l := range v.listeners { + if pr.SectionName != nil && *pr.SectionName != l.spec.Name { + continue + } + if pr.Port != nil && *pr.Port != l.spec.Port { + continue + } + out = append(out, l) + } + return out +} + +// allows reports whether a listener takes routes of this kind from this namespace. +func allows(l *listener, gw *Gateway, routeKind, routeNS string, nsLabels map[string]labels.Set) bool { + if !l.usable || l.conflicted || !contains(l.kinds, routeKind) { + return false + } + from := "Same" + var sel *metav1.LabelSelector + if l.spec.AllowedRoutes != nil && l.spec.AllowedRoutes.Namespaces != nil { + from = str(l.spec.AllowedRoutes.Namespaces.From) + if from == "" { + from = "Same" + } + sel = l.spec.AllowedRoutes.Namespaces.Selector + } + switch from { + case "All": + return true + case "Same": + return routeNS == gw.Namespace + case "Selector": + if sel == nil { + return false + } + s, err := metav1.LabelSelectorAsSelector(sel) + return err == nil && s.Matches(nsLabels[routeNS]) + } + return false +} + +func attachHTTP(r *HTTPRoute, pr ParentReference, v *gatewayView, in GatewayInput, nsLabels map[string]labels.Set, services map[string]*coreapi.Service) (metav1.Condition, metav1.Condition) { + ls := listenersFor(v, pr) + if len(ls) == 0 { + return cond(false, "NoMatchingParent", "no listener matches the parentRef's sectionName and port"), cond(true, "ResolvedRefs", "") + } + if msg := unsupportedHTTP(r); msg != "" { + return cond(false, "UnsupportedValue", msg), cond(true, "ResolvedRefs", "") + } + rules, resolved := serveRules(r, in, services) + allowed, hostOK := false, false + for _, l := range ls { + if !allows(l, v.gw, kindHTTPRoute, r.Namespace, nsLabels) { + continue + } + allowed = true + hosts, ok := narrowHostnames(str(l.spec.Hostname), r.Spec.Hostnames) + if !ok { + continue + } + hostOK = true + l.attached++ + l.http = append(l.http, ServeRoute{Key: objKey(&r.ObjectMeta), Created: r.CreationTimestamp.Time, Hostnames: hosts, Rules: rules}) + } + switch { + case !allowed: + return cond(false, "NotAllowedByListeners", "no listener the parentRef names takes HTTPRoutes from this namespace"), resolved + case !hostOK: + return cond(false, "NoMatchingListenerHostname", "the route's hostnames intersect no listener's"), resolved + } + return cond(true, "Accepted", ""), resolved +} + +func attachTCP(r *TCPRoute, pr ParentReference, v *gatewayView, in GatewayInput, nsLabels map[string]labels.Set, services map[string]*coreapi.Service) (metav1.Condition, metav1.Condition) { + ls := listenersFor(v, pr) + if len(ls) == 0 { + return cond(false, "NoMatchingParent", "no listener matches the parentRef's sectionName and port"), cond(true, "ResolvedRefs", "") + } + resolved := cond(true, "ResolvedRefs", "") + for _, rule := range r.Spec.Rules { + for _, b := range rule.BackendRefs { + if _, why, reason := resolveBackend(b, kindTCPRoute, r.Namespace, in, services); why != "" { + resolved = cond(false, reason, why) + } + } + } + allowed, attached := false, false + busy := "" + for _, l := range ls { + if !allows(l, v.gw, kindTCPRoute, r.Namespace, nsLabels) { + continue + } + allowed = true + if l.tcpRoute != nil { + // a TCP listener forwards one route: the oldest attached keeps it + busy = objKey(&l.tcpRoute.ObjectMeta) + continue + } + l.tcpRoute = r + l.attached++ + attached = true + } + switch { + case !allowed: + return cond(false, "NotAllowedByListeners", "no listener the parentRef names takes TCPRoutes from this namespace"), resolved + case !attached: + return cond(false, "NotAllowedByListeners", fmt.Sprintf("the listener already forwards TCPRoute %s", busy)), resolved + } + return cond(true, "Accepted", ""), resolved +} + +// unsupportedHTTP names what in an HTTPRoute this agent does not implement ("" if nothing). +func unsupportedHTTP(r *HTTPRoute) string { + okFilter := func(fs []HTTPRouteFilter) string { + for _, f := range fs { + switch f.Type { + case filterRequestHeaderModifier, filterResponseHeaderModifier, filterRequestRedirect, filterURLRewrite: + default: + return fmt.Sprintf("filter %s is not supported", f.Type) + } + } + return "" + } + for _, rule := range r.Spec.Rules { + for _, m := range rule.Matches { + if m.Path != nil && m.Path.Type != nil && *m.Path.Type != pathExact && *m.Path.Type != pathPrefix { + return fmt.Sprintf("path match type %s is not supported", *m.Path.Type) + } + for _, h := range m.Headers { + if t := str(h.Type); t != "" && t != "Exact" { + return fmt.Sprintf("header match type %s is not supported", t) + } + } + for _, q := range m.QueryParams { + if t := str(q.Type); t != "" && t != "Exact" { + return fmt.Sprintf("query parameter match type %s is not supported", t) + } + } + } + if msg := okFilter(rule.Filters); msg != "" { + return msg + } + for _, b := range rule.BackendRefs { + if msg := okFilter(b.Filters); msg != "" { + return msg + } + } + } + return "" +} + +// serveRules resolves an HTTPRoute's rules into what the proxy runs, and whether all its backends resolved. +func serveRules(r *HTTPRoute, in GatewayInput, services map[string]*coreapi.Service) ([]ServeRule, metav1.Condition) { + resolved := cond(true, "ResolvedRefs", "") + var rules []ServeRule + for _, rule := range r.Spec.Rules { + sr := ServeRule{Filters: rule.Filters} + for _, m := range rule.Matches { + if m.Path == nil { + m.Path = &HTTPPathMatch{} + } + if m.Path.Type == nil { + t := pathPrefix + m.Path.Type = &t + } + if m.Path.Value == nil { + v := "/" + m.Path.Value = &v + } + sr.Matches = append(sr.Matches, m) + } + if len(sr.Matches) == 0 { + t, v := pathPrefix, "/" + sr.Matches = []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: &t, Value: &v}}} + } + for _, b := range rule.BackendRefs { + eps, why, reason := resolveBackend(b.BackendRef, kindHTTPRoute, r.Namespace, in, services) + if why != "" { + resolved = cond(false, reason, why) + sr.Broken = true + continue + } + w := int32(1) + if b.Weight != nil { + w = *b.Weight + } + sr.Backends = append(sr.Backends, ServeBackend{ + Key: fmt.Sprintf("%s/%s:%d", ns(b.Namespace, r.Namespace), b.Name, *b.Port), Weight: w, Endpoints: eps, Filters: b.Filters, + }) + } + rules = append(rules, sr) + } + return rules, resolved +} + +// resolveBackend finds a backendRef's ready endpoints, or why it cannot be used and the reason to report. +func resolveBackend(b BackendRef, routeKind, routeNS string, in GatewayInput, services map[string]*coreapi.Service) ([]Endpoint, string, string) { + if group(b.Group, "") != "" || kind(b.Kind, kindService) != kindService { + return nil, fmt.Sprintf("backend %s %s is not a Service", kind(b.Kind, kindService), b.Name), "InvalidKind" + } + target := ns(b.Namespace, routeNS) + if target != routeNS && !granted(in.Grants, GatewayGroup, routeKind, routeNS, "", kindService, target, b.Name) { + return nil, fmt.Sprintf("no ReferenceGrant in %s lets %ss in %s use Service %s", target, routeKind, routeNS, b.Name), "RefNotPermitted" + } + s := services[target+"/"+b.Name] + if s == nil { + return nil, fmt.Sprintf("Service %s/%s not found", target, b.Name), "BackendNotFound" + } + if b.Port == nil { + return nil, fmt.Sprintf("backend Service %s/%s needs a port", target, b.Name), "BackendNotFound" + } + for _, p := range ports(s, in.Slices) { + if p.Port == *b.Port && p.Protocol == coreapi.ProtocolTCP { + return p.Endpoints, "", "" + } + } + return nil, fmt.Sprintf("Service %s/%s has no TCP port %d", target, b.Name, *b.Port), "BackendNotFound" +} + +// tcpEndpoints is the union of a TCPRoute's resolved backends' endpoints (the kernel forward does not weigh them). +func tcpEndpoints(r *TCPRoute, in GatewayInput, services map[string]*coreapi.Service) []Endpoint { + seen := map[string]bool{} + var out []Endpoint + for _, rule := range r.Spec.Rules { + for _, b := range rule.BackendRefs { + if b.Weight != nil && *b.Weight == 0 { + continue + } + eps, why, _ := resolveBackend(b, kindTCPRoute, r.Namespace, in, services) + if why != "" { + continue + } + for _, e := range eps { + if !seen[e.String()] { + seen[e.String()] = true + out = append(out, e) + } + } + } + } + sort.Slice(out, func(i, j int) bool { return out[i].String() < out[j].String() }) + return out +} + +// granted reports whether a ReferenceGrant in the target namespace lets (fromGroup, fromKind) objects in fromNS +// refer to the (toGroup, toKind) object named name there. +func granted(grants []*ReferenceGrant, fromGroup, fromKind, fromNS, toGroup, toKind, toNS, name string) bool { + for _, g := range grants { + if g.Namespace != toNS { + continue + } + from := false + for _, f := range g.Spec.From { + if f.Group == fromGroup && f.Kind == fromKind && f.Namespace == fromNS { + from = true + } + } + if !from { + continue + } + for _, t := range g.Spec.To { + if t.Group == toGroup && t.Kind == toKind && (t.Name == nil || *t.Name == name) { + return true + } + } + } + return false +} + +// narrowHostnames: the route's hostnames that a listener with this hostname takes (narrowed to it), and whether +// the route attaches at all. A route without hostnames takes the listener's (an empty result: any host). +func narrowHostnames(listener string, route []string) ([]string, bool) { + if len(route) == 0 { + if listener == "" { + return nil, true + } + return []string{listener}, true + } + if listener == "" { + return append([]string(nil), route...), true + } + var out []string + for _, h := range route { + switch { + case hostMatches(listener, h): + out = append(out, h) // the route's is as narrow or narrower + case hostMatches(h, listener): + out = append(out, listener) // the listener's is narrower + } + } + return out, len(out) > 0 +} + +// hostMatches reports whether a hostname pattern (exact, or "*." for any non-empty prefix) covers name, which may +// itself be a wildcard (then the pattern must cover its whole suffix). +func hostMatches(pattern, name string) bool { + pattern, name = strings.ToLower(pattern), strings.ToLower(name) + if !strings.HasPrefix(pattern, "*.") { + return pattern == name + } + suffix := pattern[1:] // ".example.com" + if strings.HasPrefix(name, "*.") { + return strings.HasSuffix(name[1:], suffix) && len(name[1:]) >= len(suffix) + } + return strings.HasSuffix(name, suffix) && len(name) > len(suffix) +} + +// gatewayStatus is what a Gateway's status should say. +func gatewayStatus(v *gatewayView, now metav1.Time) GatewayStatus { + gw := v.gw + gen := gw.Generation + var st GatewayStatus + invalid := 0 + for _, l := range v.listeners { + if !l.usable || l.conflicted || l.unbound { + invalid++ + } + } + switch { + case v.addrErr != "": + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, false, "UnsupportedAddress", v.addrErr, gen, now)) + case invalid > 0: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, true, "ListenersNotValid", fmt.Sprintf("%d of %d listeners are not valid", invalid, len(v.listeners)), gen, now)) + default: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, true, "Accepted", "", gen, now)) + } + programmed := v.addrErr == "" && v.holder != nil + switch { + case v.addrErr != "": + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, false, "Invalid", v.addrErr, gen, now)) + case v.holder == nil: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, false, "AddressNotUsable", v.holderErr, gen, now)) + default: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, true, "Programmed", "served by "+v.holder.Name, gen, now)) + ipType := "IPAddress" + st.Addresses = []GatewayAddress{{Type: &ipType, Value: v.address.String()}} + } + for _, l := range v.listeners { + cur := currentListener(gw, l.spec.Name) + ls := ListenerStatus{Name: l.spec.Name, AttachedRoutes: l.attached, SupportedKinds: []RouteGroupKind{}} + for _, k := range l.kinds { + g := GatewayGroup + ls.SupportedKinds = append(ls.SupportedKinds, RouteGroupKind{Group: &g, Kind: k}) + } + ls.Conditions = append(ls.Conditions, l.conds...) + switch { + case !programmed: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Pending", "the Gateway is not served", gen, now)) + case !l.usable || l.conflicted: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Invalid", "the listener is not valid", gen, now)) + case l.unbound: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Invalid", "its port could not be bound", gen, now)) + default: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, true, "Programmed", "", gen, now)) + } + st.Listeners = append(st.Listeners, ls) + } + return st +} + +func currentListener(gw *Gateway, name string) []metav1.Condition { + for _, l := range gw.Status.Listeners { + if l.Name == name { + return l.Conditions + } + } + return nil +} + +func currentParent(st RouteStatus, pr ParentReference) []metav1.Condition { + for _, p := range st.Parents { + if p.ControllerName == GatewayControllerName && sameJSON(p.ParentRef, pr) { + return p.Conditions + } + } + return nil +} + +// condition builds a condition, keeping the current one's transition time while its status stays the same (so an +// unchanged status reads back equal, and nothing is rewritten). +func condition(current []metav1.Condition, typ string, ok bool, reason, message string, gen int64, now metav1.Time) metav1.Condition { + status := metav1.ConditionFalse + if ok { + status = metav1.ConditionTrue + } + c := metav1.Condition{Type: typ, Status: status, Reason: reason, Message: message, ObservedGeneration: gen, LastTransitionTime: now} + for _, o := range current { + if o.Type == typ && o.Status == status { + c.LastTransitionTime = o.LastTransitionTime + } + } + return c +} + +// cond is a condition's verdict before it is dated (attach functions decide; the caller dates). +func cond(ok bool, reason, message string) metav1.Condition { + s := metav1.ConditionFalse + if ok { + s = metav1.ConditionTrue + } + return metav1.Condition{Status: s, Reason: reason, Message: message} +} + +func sameJSON(a, b interface{}) bool { + x, err1 := json.Marshal(a) + y, err2 := json.Marshal(b) + return err1 == nil && err2 == nil && string(x) == string(y) +} + +func objKey(m *metav1.ObjectMeta) string { return m.Namespace + "/" + m.Name } + +func str(p *string) string { + if p == nil { + return "" + } + return *p +} + +func group(p *string, def string) string { + if p == nil { + return def + } + return *p +} + +func kind(p *string, def string) string { + if p == nil || *p == "" { + return def + } + return *p +} + +func ns(p *string, def string) string { + if p == nil || *p == "" { + return def + } + return *p +} + +func contains(xs []string, x string) bool { + for _, y := range xs { + if y == x { + return true + } + } + return false +} + +// MergeGateways folds what a node does for Gateways into what it does for LoadBalancer Services: one set of +// forwards, one set of holders whose replies are steered back (marks over both), and the ports it serves itself. +func MergeGateways(d Desired, g GatewayDesired, self string, nodes []Node) Desired { + d.Forwards = append(d.Forwards, g.Forwards...) + holding := map[string]bool{} + for _, h := range d.Holders { + holding[h.Node.Name] = true + } + for _, n := range g.Holding { + holding[n] = true + } + sorted := append([]Node(nil), nodes...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) + var others []Node + for _, n := range sorted { + if !holding[n.Name] || n.Name == self { + continue + } + if n.WeaveIP == nil { + d.Problems = append(d.Problems, fmt.Sprintf("node %s holds an address but has not published its weave bridge; replies to its forwards cannot reach it", n.Name)) + continue + } + others = append(others, n) + } + d.Holders = assignMarks(others) + seen := map[string]bool{} + for _, l := range g.Serve { + k := fmt.Sprintf("%s:%d", l.Address, l.Port) + if seen[k] { + continue + } + seen[k] = true + d.Accept = append(d.Accept, Accept{Address: l.Address, Port: l.Port, Comment: "gateway " + l.Gateway + " " + l.Name}) + } + d.Problems = append(d.Problems, g.Problems...) + return d +} diff --git a/lb/gateway_api.go b/lb/gateway_api.go new file mode 100644 index 00000000..38983517 --- /dev/null +++ b/lb/gateway_api.go @@ -0,0 +1,415 @@ +package lb + +import ( + "context" + "encoding/json" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/rest" +) + +// The Gateway API (gateway.networking.k8s.io/v1, as the v1.6.3 standard channel serves it): the subset this agent +// implements, as plain types. The upstream Go module needs a newer client-go than the fork's; the JSON is the +// contract, and fields this agent does not implement are simply not decoded. + +const ( + GatewayGroup = "gateway.networking.k8s.io" + GatewayVersion = "v1" + // GatewayControllerName is the spec.controllerName of the GatewayClasses this agent implements. + GatewayControllerName = "reweave.levk.dev/gateway" + + GatewayClasses = "gatewayclasses" + Gateways = "gateways" + HTTPRoutes = "httproutes" + TCPRoutes = "tcproutes" + ReferenceGrants = "referencegrants" +) + +// GatewayGroupVersion of the Gateway API. +var GatewayGroupVersion = schema.GroupVersion{Group: GatewayGroup, Version: GatewayVersion} + +type GatewayClass struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec GatewayClassSpec `json:"spec"` + Status GatewayClassStatus `json:"status,omitempty"` +} + +type GatewayClassSpec struct { + ControllerName string `json:"controllerName"` +} + +type GatewayClassStatus struct { + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +type Gateway struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec GatewaySpec `json:"spec"` + Status GatewayStatus `json:"status,omitempty"` +} + +type GatewaySpec struct { + GatewayClassName string `json:"gatewayClassName"` + Listeners []Listener `json:"listeners"` + Addresses []GatewayAddress `json:"addresses,omitempty"` +} + +// GatewayAddress is a spec or status address; Type defaults to IPAddress. +type GatewayAddress struct { + Type *string `json:"type,omitempty"` + Value string `json:"value"` +} + +type Listener struct { + Name string `json:"name"` + Hostname *string `json:"hostname,omitempty"` + Port int32 `json:"port"` + Protocol string `json:"protocol"` + TLS *ListenerTLSConfig `json:"tls,omitempty"` + AllowedRoutes *AllowedRoutes `json:"allowedRoutes,omitempty"` +} + +type ListenerTLSConfig struct { + Mode *string `json:"mode,omitempty"` + CertificateRefs []SecretObjectReference `json:"certificateRefs,omitempty"` +} + +type SecretObjectReference struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Name string `json:"name"` + Namespace *string `json:"namespace,omitempty"` +} + +type AllowedRoutes struct { + Namespaces *RouteNamespaces `json:"namespaces,omitempty"` + Kinds []RouteGroupKind `json:"kinds,omitempty"` +} + +type RouteNamespaces struct { + From *string `json:"from,omitempty"` + Selector *metav1.LabelSelector `json:"selector,omitempty"` +} + +type RouteGroupKind struct { + Group *string `json:"group,omitempty"` + Kind string `json:"kind"` +} + +type GatewayStatus struct { + Addresses []GatewayAddress `json:"addresses,omitempty"` + Conditions []metav1.Condition `json:"conditions,omitempty"` + Listeners []ListenerStatus `json:"listeners,omitempty"` +} + +type ListenerStatus struct { + Name string `json:"name"` + SupportedKinds []RouteGroupKind `json:"supportedKinds"` + AttachedRoutes int32 `json:"attachedRoutes"` + Conditions []metav1.Condition `json:"conditions"` +} + +type ParentReference struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Namespace *string `json:"namespace,omitempty"` + Name string `json:"name"` + SectionName *string `json:"sectionName,omitempty"` + Port *int32 `json:"port,omitempty"` +} + +type BackendRef struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Name string `json:"name"` + Namespace *string `json:"namespace,omitempty"` + Port *int32 `json:"port,omitempty"` + Weight *int32 `json:"weight,omitempty"` +} + +type RouteStatus struct { + Parents []RouteParentStatus `json:"parents"` +} + +type RouteParentStatus struct { + ParentRef ParentReference `json:"parentRef"` + ControllerName string `json:"controllerName"` + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +type HTTPRoute struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec HTTPRouteSpec `json:"spec"` + Status RouteStatus `json:"status,omitempty"` +} + +type HTTPRouteSpec struct { + ParentRefs []ParentReference `json:"parentRefs,omitempty"` + Hostnames []string `json:"hostnames,omitempty"` + Rules []HTTPRouteRule `json:"rules,omitempty"` +} + +type HTTPRouteRule struct { + Matches []HTTPRouteMatch `json:"matches,omitempty"` + Filters []HTTPRouteFilter `json:"filters,omitempty"` + BackendRefs []HTTPBackendRef `json:"backendRefs,omitempty"` +} + +type HTTPRouteMatch struct { + Path *HTTPPathMatch `json:"path,omitempty"` + Headers []HTTPHeaderMatch `json:"headers,omitempty"` + QueryParams []HTTPQueryParamMatch `json:"queryParams,omitempty"` + Method *string `json:"method,omitempty"` +} + +type HTTPPathMatch struct { + Type *string `json:"type,omitempty"` + Value *string `json:"value,omitempty"` +} + +type HTTPHeaderMatch struct { + Type *string `json:"type,omitempty"` + Name string `json:"name"` + Value string `json:"value"` +} + +type HTTPQueryParamMatch struct { + Type *string `json:"type,omitempty"` + Name string `json:"name"` + Value string `json:"value"` +} + +// HTTPRouteFilter: the filters this agent implements are decoded; for the others only their presence is known +// (through Type), which is enough to refuse them. +type HTTPRouteFilter struct { + Type string `json:"type"` + RequestHeaderModifier *HTTPHeaderFilter `json:"requestHeaderModifier,omitempty"` + ResponseHeaderModifier *HTTPHeaderFilter `json:"responseHeaderModifier,omitempty"` + RequestRedirect *HTTPRequestRedirectFilter `json:"requestRedirect,omitempty"` + URLRewrite *HTTPURLRewriteFilter `json:"urlRewrite,omitempty"` +} + +type HTTPHeaderFilter struct { + Set []HTTPHeader `json:"set,omitempty"` + Add []HTTPHeader `json:"add,omitempty"` + Remove []string `json:"remove,omitempty"` +} + +type HTTPHeader struct { + Name string `json:"name"` + Value string `json:"value"` +} + +type HTTPRequestRedirectFilter struct { + Scheme *string `json:"scheme,omitempty"` + Hostname *string `json:"hostname,omitempty"` + Path *HTTPPathModifier `json:"path,omitempty"` + Port *int32 `json:"port,omitempty"` + StatusCode *int `json:"statusCode,omitempty"` +} + +type HTTPURLRewriteFilter struct { + Hostname *string `json:"hostname,omitempty"` + Path *HTTPPathModifier `json:"path,omitempty"` +} + +type HTTPPathModifier struct { + Type string `json:"type"` + ReplaceFullPath *string `json:"replaceFullPath,omitempty"` + ReplacePrefixMatch *string `json:"replacePrefixMatch,omitempty"` +} + +type HTTPBackendRef struct { + BackendRef `json:",inline"` + Filters []HTTPRouteFilter `json:"filters,omitempty"` +} + +type TCPRoute struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec TCPRouteSpec `json:"spec"` + Status RouteStatus `json:"status,omitempty"` +} + +type TCPRouteSpec struct { + ParentRefs []ParentReference `json:"parentRefs,omitempty"` + Rules []TCPRouteRule `json:"rules,omitempty"` +} + +type TCPRouteRule struct { + BackendRefs []BackendRef `json:"backendRefs,omitempty"` +} + +type ReferenceGrant struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec ReferenceGrantSpec `json:"spec"` +} + +type ReferenceGrantSpec struct { + From []ReferenceGrantFrom `json:"from"` + To []ReferenceGrantTo `json:"to"` +} + +type ReferenceGrantFrom struct { + Group string `json:"group"` + Kind string `json:"kind"` + Namespace string `json:"namespace"` +} + +type ReferenceGrantTo struct { + Group string `json:"group"` + Kind string `json:"kind"` + Name *string `json:"name,omitempty"` +} + +type GatewayClassList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []GatewayClass `json:"items"` +} + +type GatewayList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []Gateway `json:"items"` +} + +type HTTPRouteList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []HTTPRoute `json:"items"` +} + +type TCPRouteList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []TCPRoute `json:"items"` +} + +type ReferenceGrantList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ReferenceGrant `json:"items"` +} + +// jsonCopy deep-copies through JSON: these types are only what the JSON says, so nothing is lost. +func jsonCopy(in, out interface{}) { + b, err := json.Marshal(in) + if err != nil { + panic(err) + } + if err := json.Unmarshal(b, out); err != nil { + panic(err) + } +} + +func (o *GatewayClass) DeepCopyObject() runtime.Object { + out := &GatewayClass{} + jsonCopy(o, out) + return out +} + +func (o *Gateway) DeepCopyObject() runtime.Object { + out := &Gateway{} + jsonCopy(o, out) + return out +} + +func (o *HTTPRoute) DeepCopyObject() runtime.Object { + out := &HTTPRoute{} + jsonCopy(o, out) + return out +} + +func (o *TCPRoute) DeepCopyObject() runtime.Object { + out := &TCPRoute{} + jsonCopy(o, out) + return out +} + +func (o *ReferenceGrant) DeepCopyObject() runtime.Object { + out := &ReferenceGrant{} + jsonCopy(o, out) + return out +} + +func (o *GatewayClassList) DeepCopyObject() runtime.Object { + out := &GatewayClassList{} + jsonCopy(o, out) + return out +} + +func (o *GatewayList) DeepCopyObject() runtime.Object { + out := &GatewayList{} + jsonCopy(o, out) + return out +} + +func (o *HTTPRouteList) DeepCopyObject() runtime.Object { + out := &HTTPRouteList{} + jsonCopy(o, out) + return out +} + +func (o *TCPRouteList) DeepCopyObject() runtime.Object { + out := &TCPRouteList{} + jsonCopy(o, out) + return out +} + +func (o *ReferenceGrantList) DeepCopyObject() runtime.Object { + out := &ReferenceGrantList{} + jsonCopy(o, out) + return out +} + +func addGatewayTypes(s *runtime.Scheme) error { + s.AddKnownTypes(GatewayGroupVersion, + &GatewayClass{}, &GatewayClassList{}, &Gateway{}, &GatewayList{}, &HTTPRoute{}, &HTTPRouteList{}, + &TCPRoute{}, &TCPRouteList{}, &ReferenceGrant{}, &ReferenceGrantList{}) + metav1.AddToGroupVersion(s, GatewayGroupVersion) + return nil +} + +// NewGatewayClient is a REST client for the Gateway API kinds this agent reads and writes the status of. +func NewGatewayClient(base *rest.Config) (*rest.RESTClient, error) { + scheme := runtime.NewScheme() + if err := addGatewayTypes(scheme); err != nil { + return nil, err + } + cfg := *base + cfg.GroupVersion = &GatewayGroupVersion + cfg.APIPath = "/apis" + cfg.ContentType = runtime.ContentTypeJSON + cfg.NegotiatedSerializer = serializer.NewCodecFactory(scheme).WithoutConversion() + if cfg.UserAgent == "" { + cfg.UserAgent = rest.DefaultKubernetesUserAgent() + } + return rest.RESTClientFor(&cfg) +} + +// patchStatus merge-patches an object's status. resourceVersion, when given, makes it conditional: a status built +// from a stale read fails instead of overwriting what someone else wrote since. +func patchStatus(ctx context.Context, c rest.Interface, resource, namespace, name, resourceVersion string, status interface{}) error { + body := map[string]interface{}{"status": status} + if resourceVersion != "" { + body["metadata"] = map[string]interface{}{"resourceVersion": resourceVersion} + } + patch, err := json.Marshal(body) + if err != nil { + return err + } + req := c.Patch(types.MergePatchType).Resource(resource).Name(name).SubResource("status").Body(patch) + if namespace != "" { + req = req.Namespace(namespace) + } + return req.Do(ctx).Error() +} diff --git a/lb/gateway_proxy.go b/lb/gateway_proxy.go new file mode 100644 index 00000000..d21889eb --- /dev/null +++ b/lb/gateway_proxy.go @@ -0,0 +1,542 @@ +package lb + +import ( + "context" + "crypto/tls" + "fmt" + "log" + "math/rand" + "net" + "net/http" + "net/http/httputil" + "net/url" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + coreapi "k8s.io/api/core/v1" + + "github.com/rajch/weave/common" +) + +// Proxy serves the HTTP and HTTPS listeners of the Gateways this node holds. TLS is terminated with the +// listener's certificate, chosen by SNI and swapped in place when its Secret changes (no restart, no rebind). +// Requests are routed by the attached HTTPRoutes in the order the spec prescribes and passed to a ready endpoint +// of the chosen backend, with the client's own address in X-Forwarded-For. Bodies are streamed, never limited. +type Proxy struct { + mu sync.Mutex + servers map[string]*portServer // by "address:port" + // listen binds an address (swappable in tests) + listen func(addr string) (net.Listener, error) + transport http.RoundTripper + certs map[string]*tls.Certificate // parsed certificates, by Secret uid@resourceVersion + rr sync.Map // backend key -> *uint64, for round robin over its endpoints +} + +func NewProxy() *Proxy { + return &Proxy{ + servers: map[string]*portServer{}, + certs: map[string]*tls.Certificate{}, + listen: func(addr string) (net.Listener, error) { return net.Listen("tcp", addr) }, + transport: &http.Transport{ + DialContext: (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext, + MaxIdleConns: 256, + MaxIdleConnsPerHost: 32, + IdleConnTimeout: 90 * time.Second, + ExpectContinueTimeout: time.Second, + }, + } +} + +// portServer is one bound address:port and what it currently serves. +type portServer struct { + addr string + https bool + ln net.Listener + srv *http.Server + cfg atomic.Pointer[portConfig] +} + +// portConfig is everything one address:port serves: its listeners, most specific hostname first. +type portConfig struct { + listeners []*liveListener +} + +type liveListener struct { + spec ServeListener + cert *tls.Certificate +} + +// Apply serves exactly these listeners: ports no longer listed are closed, new ones bound, and every served +// port's routes and certificates replaced at once. It returns, by "namespace/gateway/listener", the listeners +// it could not bind. +func (p *Proxy) Apply(listeners []ServeListener) map[string]error { + p.mu.Lock() + defer p.mu.Unlock() + failed := map[string]error{} + byAddr := map[string][]ServeListener{} + for _, l := range listeners { + a := net.JoinHostPort(l.Address.String(), strconv.Itoa(int(l.Port))) + byAddr[a] = append(byAddr[a], l) + } + for a, s := range p.servers { + if group, ok := byAddr[a]; !ok || group[0].HTTPS != s.https { + s.close() + delete(p.servers, a) + } + } + used := map[string]bool{} + for a, group := range byAddr { + cfg := p.compile(group, used) + s := p.servers[a] + if s == nil { + var err error + if s, err = p.start(a, group[0].HTTPS); err != nil { + for _, l := range group { + failed[l.Gateway+"/"+l.Name] = err + } + continue + } + p.servers[a] = s + } + s.cfg.Store(cfg) + } + // forget certificates no listener uses any more + for k := range p.certs { + if !used[k] { + delete(p.certs, k) + } + } + return failed +} + +// Close stops serving everything. +func (p *Proxy) Close() { + p.Apply(nil) +} + +func (p *Proxy) compile(group []ServeListener, used map[string]bool) *portConfig { + cfg := &portConfig{} + for _, l := range group { + ll := &liveListener{spec: l} + if l.HTTPS && l.Cert != nil { + ll.cert = p.cert(l.Cert) + used[certKey(l.Cert)] = true + } + cfg.listeners = append(cfg.listeners, ll) + } + // most specific first: exact hostnames, then wildcards (longest first), then the catch-all + sort.SliceStable(cfg.listeners, func(i, j int) bool { + return hostRank(cfg.listeners[i].spec.Hostname) > hostRank(cfg.listeners[j].spec.Hostname) + }) + return cfg +} + +func certKey(s *coreapi.Secret) string { + return string(s.UID) + "@" + s.ResourceVersion + "/" + s.Namespace + "/" + s.Name +} + +// cert parses a Secret's certificate once per version of the Secret. +func (p *Proxy) cert(s *coreapi.Secret) *tls.Certificate { + k := certKey(s) + if c, ok := p.certs[k]; ok { + return c + } + c, err := tls.X509KeyPair(s.Data[coreapi.TLSCertKey], s.Data[coreapi.TLSPrivateKeyKey]) + if err != nil { + common.Log.Warnf("lb: gateway certificate %s/%s: %v", s.Namespace, s.Name, err) + return nil + } + p.certs[k] = &c + return &c +} + +// hostRank orders listener hostnames: exact (by length) above wildcards (by length) above none. +func hostRank(h string) int { + switch { + case h == "": + return 0 + case strings.HasPrefix(h, "*."): + return 1000 + len(h) + } + return 100000 + len(h) +} + +func (p *Proxy) start(addr string, https bool) (*portServer, error) { + ln, err := p.listen(addr) + if err != nil { + return nil, err + } + s := &portServer{addr: addr, https: https, ln: ln} + s.cfg.Store(&portConfig{}) + s.srv = &http.Server{ + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { p.serve(s, w, r) }), + ReadHeaderTimeout: 30 * time.Second, + IdleTimeout: 2 * time.Minute, + // handshakes from scanners and the like would flood the log + ErrorLog: log.New(debugWriter{}, "lb: gateway "+addr+": ", 0), + } + if https { + s.srv.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, GetCertificate: s.certificate} + go func() { + if err := s.srv.ServeTLS(ln, "", ""); err != nil && err != http.ErrServerClosed { + common.Log.Errorf("lb: gateway %s: %v", addr, err) + } + }() + } else { + go func() { + if err := s.srv.Serve(ln); err != nil && err != http.ErrServerClosed { + common.Log.Errorf("lb: gateway %s: %v", addr, err) + } + }() + } + common.Log.Infof("lb: gateway serving %s (%s)", addr, map[bool]string{true: "HTTPS", false: "HTTP"}[https]) + return s, nil +} + +func (s *portServer) close() { + srv := s.srv + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(ctx) + _ = srv.Close() + }() + common.Log.Infof("lb: gateway no longer serving %s", s.addr) +} + +type debugWriter struct{} + +func (debugWriter) Write(b []byte) (int, error) { + common.Log.Debugf("%s", strings.TrimRight(string(b), "\n")) + return len(b), nil +} + +// listenerFor picks the listener a name (SNI or Host) reaches on a port, most specific hostname first. +func (c *portConfig) listenerFor(name string) *liveListener { + name = strings.ToLower(name) + for _, l := range c.listeners { + if l.spec.Hostname == "" || hostMatches(l.spec.Hostname, name) { + return l + } + } + return nil +} + +// certificate answers a TLS handshake with the certificate of the listener its SNI names. +func (s *portServer) certificate(hello *tls.ClientHelloInfo) (*tls.Certificate, error) { + cfg := s.cfg.Load() + l := cfg.listenerFor(hello.ServerName) + if l == nil || l.cert == nil { + return nil, fmt.Errorf("no listener for %q", hello.ServerName) + } + return l.cert, nil +} + +// candidate is one match of one rule that a request satisfies, with what decides its precedence. +type candidate struct { + route *ServeRoute + rule *ServeRule + ruleIdx int + // the matched hostname: characters if not a wildcard, and characters + hostExact, hostLen int + exact bool + prefixLen int + method bool + headers, query int + prefix string // the path prefix matched (for ReplacePrefixMatch) +} + +func (p *Proxy) serve(s *portServer, w http.ResponseWriter, r *http.Request) { + cfg := s.cfg.Load() + host := hostOnly(r.Host) + name := host + if r.TLS != nil && r.TLS.ServerName != "" { + name = r.TLS.ServerName + } + l := cfg.listenerFor(name) + if l == nil { + http.Error(w, "no listener for this host", http.StatusNotFound) + return + } + c := pick(l, host, r) + if c == nil { + http.Error(w, "no route", http.StatusNotFound) + return + } + for _, f := range c.rule.Filters { + if f.Type == filterRequestRedirect && f.RequestRedirect != nil { + to, code := redirect(r, f.RequestRedirect, l.spec.Port, c.prefix) + http.Redirect(w, r, to, code) + return + } + } + b := chooseBackend(c.rule.Backends) + if b == nil { + // no resolved backend to send it to: the spec's answer for unresolved backends + http.Error(w, "no usable backend", http.StatusInternalServerError) + return + } + if len(b.Endpoints) == 0 { + http.Error(w, "no ready endpoint", http.StatusServiceUnavailable) + return + } + n, _ := p.rr.LoadOrStore(b.Key, new(uint64)) + ep := b.Endpoints[int(atomic.AddUint64(n.(*uint64), 1)-1)%len(b.Endpoints)] + filters := append(append([]HTTPRouteFilter(nil), c.rule.Filters...), b.Filters...) + rp := &httputil.ReverseProxy{ + Transport: p.transport, + FlushInterval: -1, // stream as it comes (git, server-sent events) + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetURL(&url.URL{Scheme: "http", Host: ep.String()}) + pr.Out.Host = pr.In.Host // forwarded unmodified unless a filter says otherwise + // SetXForwarded replaces whatever the client sent: only this hop's view is trusted + pr.SetXForwarded() + if ip, _, err := net.SplitHostPort(pr.In.RemoteAddr); err == nil { + pr.Out.Header.Set("X-Real-IP", ip) + } + for _, f := range filters { + switch { + case f.Type == filterRequestHeaderModifier && f.RequestHeaderModifier != nil: + modifyHeaders(pr.Out.Header, f.RequestHeaderModifier) + case f.Type == filterURLRewrite && f.URLRewrite != nil: + if f.URLRewrite.Hostname != nil { + pr.Out.Host = *f.URLRewrite.Hostname + } + if f.URLRewrite.Path != nil { + pr.Out.URL.Path = rewritePath(pr.In.URL.Path, f.URLRewrite.Path, c.prefix) + pr.Out.URL.RawPath = "" + } + } + } + }, + ModifyResponse: func(resp *http.Response) error { + for _, f := range filters { + if f.Type == filterResponseHeaderModifier && f.ResponseHeaderModifier != nil { + modifyHeaders(resp.Header, f.ResponseHeaderModifier) + } + } + return nil + }, + ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { + common.Log.Debugf("lb: gateway %s -> %s: %v", s.addr, ep, err) + w.WriteHeader(http.StatusBadGateway) + }, + } + rp.ServeHTTP(w, r) +} + +// pick is the match that takes the request, by the spec's precedence: the most specific matching hostname, then +// an exact path, the longest prefix, a method, the most header matches, the most query matches, then the oldest +// route, its namespace/name, and the first rule. +func pick(l *liveListener, host string, r *http.Request) *candidate { + var best *candidate + for ri := range l.spec.Routes { + route := &l.spec.Routes[ri] + hostExact, hostLen, ok := routeHost(route, l.spec.Hostname, host) + if !ok { + continue + } + for i := range route.Rules { + rule := &route.Rules[i] + for _, m := range rule.Matches { + c, ok := matches(m, r) + if !ok { + continue + } + c.route, c.rule, c.ruleIdx, c.hostExact, c.hostLen = route, rule, i, hostExact, hostLen + if best == nil || better(c, *best) { + cc := c + best = &cc + } + } + } + } + return best +} + +func better(a, b candidate) bool { + switch { + case a.hostExact != b.hostExact: + return a.hostExact > b.hostExact + case a.hostLen != b.hostLen: + return a.hostLen > b.hostLen + case a.exact != b.exact: + return a.exact + case a.prefixLen != b.prefixLen: + return a.prefixLen > b.prefixLen + case a.method != b.method: + return a.method + case a.headers != b.headers: + return a.headers > b.headers + case a.query != b.query: + return a.query > b.query + case !a.route.Created.Equal(b.route.Created): + return a.route.Created.Before(b.route.Created) + case a.route.Key != b.route.Key: + return a.route.Key < b.route.Key + } + return a.ruleIdx < b.ruleIdx +} + +// routeHost reports whether a route answers this host on its listener, and how specifically. +func routeHost(route *ServeRoute, listenerHost, host string) (int, int, bool) { + names := route.Hostnames + if len(names) == 0 { + if listenerHost == "" { + return 0, 0, true + } + names = []string{listenerHost} + } + exact, length, ok := 0, 0, false + for _, h := range names { + if !hostMatches(h, host) { + continue + } + ok = true + e := 0 + if !strings.HasPrefix(h, "*.") { + e = len(h) + } + if e > exact || (e == exact && len(h) > length) { + exact, length = e, len(h) + } + } + return exact, length, ok +} + +func matches(m HTTPRouteMatch, r *http.Request) (candidate, bool) { + var c candidate + path, value := r.URL.Path, str(m.Path.Value) + switch str(m.Path.Type) { + case pathExact: + if path != value { + return c, false + } + c.exact = true + c.prefixLen = len(value) + default: + trimmed := strings.TrimSuffix(value, "/") + if trimmed != "" && path != trimmed && !strings.HasPrefix(path, trimmed+"/") { + return c, false + } + c.prefixLen = len(value) + c.prefix = value + } + if m.Method != nil { + if r.Method != *m.Method { + return c, false + } + c.method = true + } + for _, h := range m.Headers { + if r.Header.Get(h.Name) != h.Value { + return c, false + } + } + c.headers = len(m.Headers) + q := r.URL.Query() + for _, qp := range m.QueryParams { + if q.Get(qp.Name) != qp.Value { + return c, false + } + } + c.query = len(m.QueryParams) + return c, true +} + +// chooseBackend picks a backend by weight (nil when there is none to pick). +func chooseBackend(bs []ServeBackend) *ServeBackend { + total := int32(0) + for _, b := range bs { + if b.Weight > 0 { + total += b.Weight + } + } + if total == 0 { + return nil + } + n := rand.Int31n(total) + for i := range bs { + if bs[i].Weight <= 0 { + continue + } + if n < bs[i].Weight { + return &bs[i] + } + n -= bs[i].Weight + } + return nil +} + +// redirect is where a RequestRedirect filter sends a request, and with which status. +func redirect(r *http.Request, f *HTTPRequestRedirectFilter, listenerPort int32, prefix string) (string, int) { + scheme := "http" + if r.TLS != nil { + scheme = "https" + } + port := int(listenerPort) + if f.Scheme != nil { + // a new scheme brings its well-known port, unless a port is given + scheme = *f.Scheme + port = map[string]int{"http": 80, "https": 443}[scheme] + } + if f.Port != nil { + port = int(*f.Port) + } + host := hostOnly(r.Host) + if f.Hostname != nil { + host = *f.Hostname + } + if !(scheme == "http" && port == 80) && !(scheme == "https" && port == 443) && port != 0 { + host = net.JoinHostPort(host, strconv.Itoa(port)) + } + path := r.URL.Path + if f.Path != nil { + path = rewritePath(path, f.Path, prefix) + } + code := http.StatusFound + if f.StatusCode != nil { + code = *f.StatusCode + } + return (&url.URL{Scheme: scheme, Host: host, Path: path, RawQuery: r.URL.RawQuery}).String(), code +} + +// rewritePath applies a path modifier: the whole path, or the matched prefix replaced. +func rewritePath(path string, m *HTTPPathModifier, prefix string) string { + switch m.Type { + case "ReplaceFullPath": + return str(m.ReplaceFullPath) + case "ReplacePrefixMatch": + rest := strings.TrimPrefix(path, strings.TrimSuffix(prefix, "/")) + out := strings.TrimSuffix(str(m.ReplacePrefixMatch), "/") + rest + if !strings.HasPrefix(out, "/") { + out = "/" + out + } + return out + } + return path +} + +func modifyHeaders(h http.Header, f *HTTPHeaderFilter) { + for _, s := range f.Set { + h.Set(s.Name, s.Value) + } + for _, a := range f.Add { + h.Add(a.Name, a.Value) + } + for _, name := range f.Remove { + h.Del(name) + } +} + +// hostOnly is a Host header without its port, lowercased (the spec ignores the port when matching). +func hostOnly(h string) string { + if host, _, err := net.SplitHostPort(h); err == nil { + h = host + } + return strings.ToLower(strings.TrimSuffix(h, ".")) +} diff --git a/lb/gateway_proxy_test.go b/lb/gateway_proxy_test.go new file mode 100644 index 00000000..20d63796 --- /dev/null +++ b/lb/gateway_proxy_test.go @@ -0,0 +1,314 @@ +package lb + +import ( + "bytes" + "crypto/tls" + "crypto/x509" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + coreapi "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" +) + +// testProxy binds every requested address on loopback and remembers where. +type testProxy struct { + *Proxy + mu sync.Mutex + actual map[string]string +} + +func newTestProxy(t *testing.T) *testProxy { + tp := &testProxy{Proxy: NewProxy(), actual: map[string]string{}} + tp.listen = func(addr string) (net.Listener, error) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err == nil { + tp.mu.Lock() + tp.actual[addr] = ln.Addr().String() + tp.mu.Unlock() + } + return ln, err + } + t.Cleanup(tp.Close) + return tp +} + +func (tp *testProxy) at(port int32) string { + tp.mu.Lock() + defer tp.mu.Unlock() + return tp.actual[net.JoinHostPort("192.0.2.1", strconv.Itoa(int(port)))] +} + +// named is a backend that answers with its name and the request it saw. +func named(t *testing.T, name string) (*httptest.Server, Endpoint) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Backend", name) + n, _ := io.Copy(io.Discard, r.Body) + fmt.Fprintf(w, "%s path=%s host=%s xff=%s proto=%s xhost=%s realip=%s added=%s removed=%s body=%d", + name, r.URL.Path, r.Host, r.Header.Get("X-Forwarded-For"), r.Header.Get("X-Forwarded-Proto"), + r.Header.Get("X-Forwarded-Host"), r.Header.Get("X-Real-IP"), strings.Join(r.Header.Values("X-Added"), ","), + r.Header.Get("X-Remove-Me"), n) + })) + t.Cleanup(s.Close) + _, port, _ := net.SplitHostPort(s.Listener.Addr().String()) + p, _ := strconv.Atoi(port) + return s, Endpoint{IP: net.ParseIP("127.0.0.1").To4(), Port: int32(p)} +} + +func backend(name string, ep Endpoint) []ServeBackend { + return []ServeBackend{{Key: name, Weight: 1, Endpoints: []Endpoint{ep}}} +} + +func prefix(v string) HTTPRouteMatch { + return HTTPRouteMatch{Path: &HTTPPathMatch{Type: ptr(pathPrefix), Value: ptr(v)}} +} + +func exact(v string) HTTPRouteMatch { + return HTTPRouteMatch{Path: &HTTPPathMatch{Type: ptr(pathExact), Value: ptr(v)}} +} + +func get(t *testing.T, url, host string, hdr map[string]string) (int, string, http.Header) { + t.Helper() + req, _ := http.NewRequest("GET", url, nil) + if host != "" { + req.Host = host + } + for k, v := range hdr { + req.Header.Set(k, v) + } + c := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + resp, err := c.Do(req) + if err != nil { + t.Fatalf("GET %s: %v", url, err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(b), resp.Header +} + +func TestProxyRoutesByPrecedence(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + _, b := named(t, "B") + _, c := named(t, "C") + _, d := named(t, "D") + _, e := named(t, "E") + t0 := time.Date(2026, 10, 14, 1, 0, 0, 0, time.UTC) + routes := []ServeRoute{ + {Key: "ns/first", Created: t0, Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{exact("/exact")}, Backends: backend("A", a)}, + {Matches: []HTTPRouteMatch{prefix("/api")}, Backends: backend("B", b)}, + {Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("C", c)}, + }}, + {Key: "ns/second", Created: t0.Add(time.Minute), Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/api/v2")}, Backends: backend("D", d)}, + {Matches: []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: ptr(pathPrefix), Value: ptr("/api")}, + Headers: []HTTPHeaderMatch{{Name: "X-Ver", Value: "2"}}}}, Backends: backend("E", e)}, + }}, + } + if f := tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8080, Routes: routes}}); len(f) != 0 { + t.Fatal(f) + } + base := "http://" + tp.at(8080) + for path, want := range map[string]string{ + "/exact": "A", "/exactly": "C", "/api/x": "B", "/api": "B", "/api/v2/y": "D", "/apix": "C", "/": "C", + } { + if _, body, _ := get(t, base+path, "example.com", nil); !strings.HasPrefix(body, want+" ") { + t.Errorf("%s went to %q, want %s", path, body, want) + } + } + // same prefix length: the match with more headers takes it, though its route is younger + if _, body, _ := get(t, base+"/api/x", "example.com", map[string]string{"X-Ver": "2"}); !strings.HasPrefix(body, "E ") { + t.Errorf("header match lost: %q", body) + } +} + +func TestProxyHostPrecedence(t *testing.T) { + tp := newTestProxy(t) + _, w := named(t, "W") + _, x := named(t, "X") + routes := []ServeRoute{ + {Key: "ns/wild", Hostnames: []string{"*.example.com"}, Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("W", w)}}}, + {Key: "ns/exact", Hostnames: []string{"a.example.com"}, Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("X", x)}}}, + } + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8081, Routes: routes}}) + base := "http://" + tp.at(8081) + for host, want := range map[string]string{"a.example.com": "X", "A.Example.com:8081": "X", "b.example.com": "W"} { + if _, body, _ := get(t, base+"/", host, nil); !strings.HasPrefix(body, want+" ") { + t.Errorf("host %s went to %q, want %s", host, body, want) + } + } + if code, _, _ := get(t, base+"/", "example.org", nil); code != http.StatusNotFound { + t.Errorf("a host no route names: %d, want 404", code) + } +} + +func certSecret(t *testing.T, name, rv string) *coreapi.Secret { + s := tlsSecret(t, "gateway-system", strings.ReplaceAll(name, ".", "-"), name) + s.UID = types.UID("uid-" + name) + s.ResourceVersion = rv + return s +} + +func dialTLS(t *testing.T, addr, sni string) (*x509.Certificate, error) { + conn, err := tls.Dial("tcp", addr, &tls.Config{ServerName: sni, InsecureSkipVerify: true}) + if err != nil { + return nil, err + } + defer conn.Close() + return conn.ConnectionState().PeerCertificates[0], nil +} + +func TestProxyTLSBySNIAndSwapsCertificates(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + _, b := named(t, "B") + listeners := func(certA *coreapi.Secret) []ServeListener { + return []ServeListener{ + {Gateway: "ns/gw", Name: "a", Address: net.ParseIP("192.0.2.1"), Port: 8443, HTTPS: true, Hostname: "a.test", Cert: certA, + Routes: []ServeRoute{{Key: "ns/a", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}}}}}, + {Gateway: "ns/gw", Name: "b", Address: net.ParseIP("192.0.2.1"), Port: 8443, HTTPS: true, Hostname: "b.test", Cert: certSecret(t, "b.test", "1"), + Routes: []ServeRoute{{Key: "ns/b", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("B", b)}}}}}, + } + } + first := certSecret(t, "a.test", "1") + if f := tp.Apply(listeners(first)); len(f) != 0 { + t.Fatal(f) + } + addr := tp.at(8443) + ca, err := dialTLS(t, addr, "a.test") + if err != nil || ca.Subject.CommonName != "a.test" { + t.Fatalf("SNI a.test: %v %v", ca, err) + } + cb, err := dialTLS(t, addr, "b.test") + if err != nil || cb.Subject.CommonName != "b.test" { + t.Fatalf("SNI b.test: %v %v", cb, err) + } + if _, err := dialTLS(t, addr, "c.test"); err == nil { + t.Fatal("a name no listener has must not get a certificate") + } + // requests go by listener, over HTTP/2 when the client offers it + client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, ServerName: "b.test"}, ForceAttemptHTTP2: true}} + req, _ := http.NewRequest("GET", "https://"+addr+"/", nil) + req.Host = "b.test" + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !strings.HasPrefix(string(body), "B ") || !strings.Contains(string(body), "proto=https") || resp.ProtoMajor != 2 { + t.Fatalf("over HTTPS to b.test: %s %q", resp.Proto, body) + } + // a renewed Secret: new handshakes get the new certificate on the same, still-bound port + renewed := certSecret(t, "a.test", "2") + tp.Apply(listeners(renewed)) + if tp.at(8443) != addr { + t.Fatal("the port was rebound for a certificate change") + } + cn, err := dialTLS(t, addr, "a.test") + if err != nil || cn.SerialNumber.Cmp(ca.SerialNumber) == 0 { + t.Fatalf("still the old certificate after the swap: %v", err) + } +} + +func TestProxyRedirectsAndForwards(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + routes := []ServeRoute{{Key: "ns/r", Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/old")}, Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Scheme: ptr("https"), StatusCode: ptr(301)}}}}, + {Matches: []HTTPRouteMatch{prefix("/moved")}, Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Path: &HTTPPathModifier{Type: "ReplacePrefixMatch", ReplacePrefixMatch: ptr("/new")}}}}}, + {Matches: []HTTPRouteMatch{prefix("/v1")}, Backends: backend("A", a), Filters: []HTTPRouteFilter{ + {Type: filterRequestHeaderModifier, RequestHeaderModifier: &HTTPHeaderFilter{ + Set: []HTTPHeader{{Name: "X-Set", Value: "s"}}, Add: []HTTPHeader{{Name: "X-Added", Value: "two"}}, Remove: []string{"X-Remove-Me"}}}, + {Type: filterResponseHeaderModifier, ResponseHeaderModifier: &HTTPHeaderFilter{Set: []HTTPHeader{{Name: "X-Resp", Value: "r"}}}}, + {Type: filterURLRewrite, URLRewrite: &HTTPURLRewriteFilter{Path: &HTTPPathModifier{Type: "ReplacePrefixMatch", ReplacePrefixMatch: ptr("/v2")}}}, + }}, + {Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}, + }}} + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 80, Routes: routes}}) + base := "http://" + tp.at(80) + code, _, h := get(t, base+"/old/page?q=1", "git.levk.dev", nil) + if code != http.StatusMovedPermanently || h.Get("Location") != "https://git.levk.dev/old/page?q=1" { + t.Fatalf("scheme redirect: %d %q", code, h.Get("Location")) + } + code, _, h = get(t, base+"/moved/x", "git.levk.dev", nil) + if code != http.StatusFound || h.Get("Location") != "http://git.levk.dev/new/x" { + t.Fatalf("prefix redirect: %d %q", code, h.Get("Location")) + } + // the client's own claims about who it is are not passed on; the backend sees this hop's view + _, body, _ := get(t, base+"/", "git.levk.dev", map[string]string{"X-Forwarded-For": "6.6.6.6", "X-Real-IP": "6.6.6.6"}) + if !strings.Contains(body, "host=git.levk.dev") || !strings.Contains(body, "xff=127.0.0.1 ") || + !strings.Contains(body, "proto=http ") || !strings.Contains(body, "xhost=git.levk.dev") || !strings.Contains(body, "realip=127.0.0.1") { + t.Fatalf("forwarded headers: %q", body) + } + _, body, h = get(t, base+"/v1/thing", "git.levk.dev", map[string]string{"X-Added": "one", "X-Remove-Me": "x"}) + if !strings.Contains(body, "path=/v2/thing") || !strings.Contains(body, "added=one,two") || !strings.Contains(body, "removed= ") || h.Get("X-Resp") != "r" { + t.Fatalf("filters: %q %v", body, h) + } +} + +func TestProxyAnswersWhatItCannotRoute(t *testing.T) { + tp := newTestProxy(t) + routes := []ServeRoute{{Key: "ns/r", Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/broken")}, Broken: true}, + {Matches: []HTTPRouteMatch{prefix("/empty")}, Backends: []ServeBackend{{Key: "ns/svc:80", Weight: 1}}}, + }}} + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8082, Routes: routes}}) + base := "http://" + tp.at(8082) + for path, want := range map[string]int{"/broken": 500, "/empty": 503, "/nothing": 404} { + if code, _, _ := get(t, base+path, "x", nil); code != want { + t.Errorf("%s: %d, want %d", path, code, want) + } + } +} + +func TestProxyStreamsLargeBodies(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8083, + Routes: []ServeRoute{{Key: "ns/r", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}}}}}}) + const size = 32 << 20 + resp, err := http.Post("http://"+tp.at(8083)+"/upload", "application/octet-stream", bytes.NewReader(make([]byte, size))) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !strings.Contains(string(body), fmt.Sprintf("body=%d", size)) { + t.Fatalf("upload: %q", body) + } +} + +func TestProxyBindFailureAndClose(t *testing.T) { + tp := newTestProxy(t) + tp.listen = func(addr string) (net.Listener, error) { return nil, fmt.Errorf("address in use") } + f := tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8084}}) + if f["ns/gw/http"] == nil { + t.Fatalf("bind failure not reported: %v", f) + } + tp2 := newTestProxy(t) + tp2.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8085}}) + addr := tp2.at(8085) + tp2.Apply(nil) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + c, err := net.DialTimeout("tcp", addr, 200*time.Millisecond) + if err != nil { + return + } + c.Close() + time.Sleep(50 * time.Millisecond) + } + t.Fatal("a port no longer served still accepts connections") +} diff --git a/lb/gateway_test.go b/lb/gateway_test.go new file mode 100644 index 00000000..e1176ef1 --- /dev/null +++ b/lb/gateway_test.go @@ -0,0 +1,576 @@ +package lb + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net" + "strings" + "testing" + "time" + + coreapi "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +var gwNow = metav1.NewTime(time.Date(2026, 10, 14, 4, 0, 0, 0, time.UTC)) + +// selfSigned is a certificate and key for the given names, PEM-encoded. +func selfSigned(t *testing.T, names ...string) ([]byte, []byte) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(time.Now().UnixNano()), Subject: pkix.Name{CommonName: names[0]}, DNSNames: names, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + kb, _ := x509.MarshalECPrivateKey(key) + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: kb}) +} + +func tlsSecret(t *testing.T, namespace, name string, names ...string) *coreapi.Secret { + crt, key := selfSigned(t, names...) + return &coreapi.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, Type: coreapi.SecretTypeTLS, + Data: map[string][]byte{coreapi.TLSCertKey: crt, coreapi.TLSPrivateKeyKey: key}} +} + +func gwNodes() []Node { + return []Node{ + {Name: "cp", Addresses: []net.IP{net.ParseIP("86.107.168.24").To4()}, WeaveIP: net.ParseIP("10.32.0.1").To4(), WeaveMAC: mustMAC("02:00:00:00:00:01")}, + {Name: "w15", Addresses: []net.IP{net.ParseIP("86.107.168.15").To4()}, WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mustMAC("02:00:00:00:00:02")}, + } +} + +func mustMAC(s string) net.HardwareAddr { + m, _ := net.ParseMAC(s) + return m +} + +func ourClass() *GatewayClass { + return &GatewayClass{ObjectMeta: metav1.ObjectMeta{Name: "reweave", Generation: 1}, Spec: GatewayClassSpec{ControllerName: GatewayControllerName}} +} + +// giteaGateway is the shape Gitea's documentation prescribes: HTTPS with a certificate, plain HTTP, and SSH over TCP. +func giteaGateway() *Gateway { + return &Gateway{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gateway-system", Name: "shared", Generation: 3}, + Spec: GatewaySpec{ + GatewayClassName: "reweave", + Addresses: []GatewayAddress{{Value: "86.107.168.15"}}, + Listeners: []Listener{ + {Name: "https-gitea", Port: 443, Protocol: ProtocolHTTPS, Hostname: ptr("git.levk.dev"), + TLS: &ListenerTLSConfig{CertificateRefs: []SecretObjectReference{{Name: "git-levk-dev"}}}, + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("Selector"), Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": "gitea"}}}}}, + {Name: "http", Port: 80, Protocol: ProtocolHTTP, Hostname: ptr("git.levk.dev"), + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("All")}}}, + {Name: "ssh", Port: 22, Protocol: ProtocolTCP, + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("All")}}}, + }, + }, + } +} + +func giteaService() *coreapi.Service { + return &coreapi.Service{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea"}, Spec: coreapi.ServiceSpec{ + Ports: []coreapi.ServicePort{{Name: "http", Port: 3000}, {Name: "ssh", Port: 22}}, + }} +} + +func giteaSlice() *discovery.EndpointSlice { + return &discovery.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-abc", Labels: map[string]string{serviceNameLabel: "gitea"}}, + AddressType: discovery.AddressTypeIPv4, + Ports: []discovery.EndpointPort{{Name: ptr("http"), Port: ptr(int32(3000))}, {Name: ptr("ssh"), Port: ptr(int32(22))}}, + Endpoints: []discovery.Endpoint{{Addresses: []string{"10.44.0.12"}, Conditions: discovery.EndpointConditions{Ready: ptr(true)}}}, + } +} + +func giteaRoutes() ([]*HTTPRoute, []*TCPRoute) { + t0 := metav1.NewTime(time.Date(2026, 10, 14, 1, 0, 0, 0, time.UTC)) + https := &HTTPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea", Generation: 1, CreationTimestamp: t0}, + Spec: HTTPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("https-gitea")}}, + Hostnames: []string{"git.levk.dev"}, + Rules: []HTTPRouteRule{{BackendRefs: []HTTPBackendRef{{BackendRef: BackendRef{Name: "gitea", Port: ptr(int32(3000))}}}}}, + }, + } + redirect := &HTTPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-redirect", Generation: 1, CreationTimestamp: t0}, + Spec: HTTPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("http")}}, + Rules: []HTTPRouteRule{{Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Scheme: ptr("https"), StatusCode: ptr(301)}}}}}, + }, + } + ssh := &TCPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-ssh", Generation: 1, CreationTimestamp: t0}, + Spec: TCPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("ssh")}}, + Rules: []TCPRouteRule{{BackendRefs: []BackendRef{{Name: "gitea", Port: ptr(int32(22))}}}}, + }, + } + return []*HTTPRoute{https, redirect}, []*TCPRoute{ssh} +} + +func giteaInput(t *testing.T) GatewayInput { + hr, tr := giteaRoutes() + return GatewayInput{ + Classes: []*GatewayClass{ourClass()}, Gateways: []*Gateway{giteaGateway()}, HTTPRoutes: hr, TCPRoutes: tr, + Namespaces: []*coreapi.Namespace{ + {ObjectMeta: metav1.ObjectMeta{Name: "gitea", Labels: map[string]string{"kubernetes.io/metadata.name": "gitea"}}}, + {ObjectMeta: metav1.ObjectMeta{Name: "gateway-system", Labels: map[string]string{"kubernetes.io/metadata.name": "gateway-system"}}}, + }, + Services: []*coreapi.Service{giteaService()}, Slices: []*discovery.EndpointSlice{giteaSlice()}, + Secrets: []*coreapi.Secret{tlsSecret(t, "gateway-system", "git-levk-dev", "git.levk.dev")}, + SecretNamespaces: map[string]bool{"gateway-system": true}, + Now: gwNow, + } +} + +func condOf(cs []metav1.Condition, typ string) metav1.Condition { + for _, c := range cs { + if c.Type == typ { + return c + } + } + return metav1.Condition{} +} + +func TestGatewayServedByItsHolder(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("w15", gwNodes(), in) + if len(d.Problems) != 0 { + t.Fatalf("problems: %v", d.Problems) + } + // HTTPS and HTTP served by the agent; SSH forwarded in the kernel to Gitea's pod with the client address intact + if len(d.Serve) != 2 { + t.Fatalf("want the https and http listeners served, got %+v", d.Serve) + } + https := d.Serve[0] + if !https.HTTPS || https.Port != 443 || https.Hostname != "git.levk.dev" || https.Cert == nil || len(https.Routes) != 1 { + t.Fatalf("https listener: %+v", https) + } + be := https.Routes[0].Rules[0].Backends + if len(be) != 1 || be[0].Key != "gitea/gitea:3000" || len(be[0].Endpoints) != 1 || be[0].Endpoints[0].String() != "10.44.0.12:3000" { + t.Fatalf("https backends: %+v", be) + } + if len(d.Forwards) != 1 || d.Forwards[0].Ports[0].Port != 22 || d.Forwards[0].Ports[0].Endpoints[0].String() != "10.44.0.12:22" { + t.Fatalf("ssh forward: %+v", d.Forwards) + } + if len(d.Holding) != 1 || d.Holding[0] != "w15" { + t.Fatalf("holding: %v", d.Holding) + } + // the holder writes the Gateway's status: accepted, programmed, its address, each listener with one route + if len(d.Gateways) != 1 { + t.Fatalf("gateway status writes: %+v", d.Gateways) + } + st := d.Gateways[0].Status + if c := condOf(st.Conditions, condProgrammed); c.Status != metav1.ConditionTrue || c.ObservedGeneration != 3 { + t.Fatalf("programmed: %+v", c) + } + if len(st.Addresses) != 1 || st.Addresses[0].Value != "86.107.168.15" { + t.Fatalf("addresses: %+v", st.Addresses) + } + for _, l := range st.Listeners { + if l.AttachedRoutes != 1 || condOf(l.Conditions, condProgrammed).Status != metav1.ConditionTrue { + t.Fatalf("listener %s: %+v", l.Name, l) + } + } + // route statuses come from the first node by name, not the holder + if len(d.Routes) != 0 || len(d.Classes) != 0 { + t.Fatalf("w15 wrote route or class statuses: %+v %+v", d.Routes, d.Classes) + } +} + +func TestGatewayElsewhere(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("cp", gwNodes(), in) + if len(d.Serve) != 0 || len(d.Forwards) != 0 || len(d.Gateways) != 0 { + t.Fatalf("cp serves or writes what w15 holds: %+v %+v %+v", d.Serve, d.Forwards, d.Gateways) + } + // but steers replies of w15's forwards back to it, and writes class and route statuses (first by name) + if len(d.Holding) != 1 || d.Holding[0] != "w15" { + t.Fatalf("holding: %v", d.Holding) + } + if len(d.Classes) != 1 || condOf(d.Classes[0].Status.Conditions, condAccepted).Status != metav1.ConditionTrue { + t.Fatalf("class status: %+v", d.Classes) + } + if len(d.Routes) != 3 { + t.Fatalf("want 3 route statuses, got %+v", d.Routes) + } + for _, r := range d.Routes { + if len(r.Status.Parents) != 1 || condOf(r.Status.Parents[0].Conditions, condAccepted).Status != metav1.ConditionTrue || + condOf(r.Status.Parents[0].Conditions, condResolvedRefs).Status != metav1.ConditionTrue { + t.Fatalf("route %s/%s status: %+v", r.Namespace, r.Name, r.Status) + } + } +} + +func TestGatewayStatusIsWrittenOnce(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("w15", gwNodes(), in) + in.Gateways[0].Status = d.Gateways[0].Status + in.Now = metav1.NewTime(gwNow.Add(time.Hour)) // later passes see the status they wrote + if again := ComputeGateways("w15", gwNodes(), in); len(again.Gateways) != 0 { + t.Fatalf("unchanged status rewritten: %+v", again.Gateways) + } + c := ComputeGateways("cp", gwNodes(), in) + for _, r := range c.Routes { + for _, hr := range in.HTTPRoutes { + if hr.Name == r.Name { + hr.Status = r.Status + } + } + for _, tr := range in.TCPRoutes { + if tr.Name == r.Name { + tr.Status = r.Status + } + } + } + in.Classes[0].Status = c.Classes[0].Status + if again := ComputeGateways("cp", gwNodes(), in); len(again.Routes) != 0 || len(again.Classes) != 0 { + t.Fatalf("unchanged route or class status rewritten: %+v %+v", again.Routes, again.Classes) + } +} + +func TestGatewayFailsClosedWithoutHolder(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Addresses[0].Value = "198.51.100.7" + for _, self := range []string{"cp", "w15"} { + d := ComputeGateways(self, gwNodes(), in) + if len(d.Serve) != 0 || len(d.Forwards) != 0 || len(d.Holding) != 0 { + t.Fatalf("%s serves an address nobody holds: %+v", self, d) + } + } + d := ComputeGateways("cp", gwNodes(), in) + if len(d.Gateways) != 1 { + t.Fatalf("the first node by name must write the failure: %+v", d.Gateways) + } + c := condOf(d.Gateways[0].Status.Conditions, condProgrammed) + if c.Status != metav1.ConditionFalse || c.Reason != "AddressNotUsable" || len(d.Gateways[0].Status.Addresses) != 0 { + t.Fatalf("status: %+v", d.Gateways[0].Status) + } +} + +func TestGatewayAddressMustBeOneIPv4(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Addresses = nil + d := ComputeGateways("cp", gwNodes(), in) + if c := condOf(d.Gateways[0].Status.Conditions, condAccepted); c.Status != metav1.ConditionFalse || c.Reason != "UnsupportedAddress" { + t.Fatalf("accepted: %+v", c) + } +} + +func TestListenerCertificateReferences(t *testing.T) { + cases := map[string]struct { + mutate func(in *GatewayInput) + reason string + }{ + "missing Secret": {func(in *GatewayInput) { in.Secrets = nil }, "InvalidCertificateRef"}, + "other namespace without grant": {func(in *GatewayInput) { + in.Gateways[0].Spec.Listeners[0].TLS.CertificateRefs[0].Namespace = ptr("gitea") + in.SecretNamespaces["gitea"] = true + }, "RefNotPermitted"}, + "namespace the agent cannot read": {func(in *GatewayInput) { + in.SecretNamespaces = map[string]bool{} + }, "RefNotPermitted"}, + "not a certificate": {func(in *GatewayInput) { + in.Secrets[0].Data[coreapi.TLSCertKey] = []byte("nope") + }, "InvalidCertificateRef"}, + } + for name, c := range cases { + in := giteaInput(t) + c.mutate(&in) + d := ComputeGateways("w15", gwNodes(), in) + var https ListenerStatus + for _, l := range d.Gateways[0].Status.Listeners { + if l.Name == "https-gitea" { + https = l + } + } + if r := condOf(https.Conditions, condResolvedRefs); r.Status != metav1.ConditionFalse || r.Reason != c.reason { + t.Errorf("%s: resolvedRefs %+v", name, r) + } + if p := condOf(https.Conditions, condProgrammed); p.Status != metav1.ConditionFalse { + t.Errorf("%s: an HTTPS listener without its certificate must not be programmed", name) + } + for _, s := range d.Serve { + if s.HTTPS { + t.Errorf("%s: HTTPS listener served without a certificate", name) + } + } + if a := condOf(d.Gateways[0].Status.Conditions, condAccepted); a.Reason != "ListenersNotValid" || a.Status != metav1.ConditionTrue { + t.Errorf("%s: gateway accepted %+v", name, a) + } + } +} + +func TestCrossNamespaceCertificateWithGrant(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Listeners[0].TLS.CertificateRefs[0].Namespace = ptr("gitea") + in.Secrets = []*coreapi.Secret{tlsSecret(t, "gitea", "git-levk-dev", "git.levk.dev")} + in.SecretNamespaces = map[string]bool{"gitea": true} + in.Grants = []*ReferenceGrant{{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gw"}, Spec: ReferenceGrantSpec{ + From: []ReferenceGrantFrom{{Group: GatewayGroup, Kind: kindGateway, Namespace: "gateway-system"}}, + To: []ReferenceGrantTo{{Group: "", Kind: kindSecret}}, + }}} + d := ComputeGateways("w15", gwNodes(), in) + if len(d.Serve) != 2 || d.Serve[0].Cert == nil { + t.Fatalf("a granted certificate in another namespace must be served: %+v", d.Serve) + } +} + +func TestListenerConflicts(t *testing.T) { + in := giteaInput(t) + l := in.Gateways[0].Spec.Listeners + in.Gateways[0].Spec.Listeners = append(l, + Listener{Name: "ssh2", Port: 22, Protocol: ProtocolTCP}, // two TCP listeners on 22 + Listener{Name: "http-on-443", Port: 443, Protocol: ProtocolHTTP}, // HTTP and HTTPS on 443 + ) + d := ComputeGateways("w15", gwNodes(), in) + want := map[string]string{"ssh": "HostnameConflict", "ssh2": "HostnameConflict", "https-gitea": "ProtocolConflict", "http-on-443": "ProtocolConflict", "http": ""} + for _, ls := range d.Gateways[0].Status.Listeners { + c := condOf(ls.Conditions, condConflicted) + if want[ls.Name] == "" { + if c.Status != metav1.ConditionFalse { + t.Errorf("%s: conflicted %+v", ls.Name, c) + } + continue + } + if c.Status != metav1.ConditionTrue || c.Reason != want[ls.Name] { + t.Errorf("%s: want conflicted %s, got %+v", ls.Name, want[ls.Name], c) + } + } + if len(d.Forwards) != 0 { + t.Fatalf("conflicted TCP listeners forwarded: %+v", d.Forwards) + } + if len(d.Serve) != 1 || d.Serve[0].Name != "http" { + t.Fatalf("only the plain http listener on 80 is unconflicted: %+v", d.Serve) + } +} + +func TestListenerPortTakenByLoadBalancerService(t *testing.T) { + in := giteaInput(t) + class := Class + in.Services = append(in.Services, &coreapi.Service{ObjectMeta: metav1.ObjectMeta{Namespace: "mail", Name: "smtp"}, Spec: coreapi.ServiceSpec{ + Type: coreapi.ServiceTypeLoadBalancer, LoadBalancerClass: &class, LoadBalancerIP: "86.107.168.15", + Ports: []coreapi.ServicePort{{Port: 22}}, + }}) + d := ComputeGateways("w15", gwNodes(), in) + for _, ls := range d.Gateways[0].Status.Listeners { + if ls.Name == "ssh" { + if c := condOf(ls.Conditions, condAccepted); c.Status != metav1.ConditionFalse || c.Reason != "PortUnavailable" { + t.Fatalf("ssh accepted: %+v", c) + } + } + } + if len(d.Forwards) != 0 { + t.Fatalf("a listener on a port a Service forwards must not be forwarded: %+v", d.Forwards) + } +} + +func TestRouteAttachment(t *testing.T) { + cases := map[string]struct { + mutate func(in *GatewayInput) + route string + reason string + }{ + "unknown section": {func(in *GatewayInput) { in.HTTPRoutes[0].Spec.ParentRefs[0].SectionName = ptr("nope") }, "gitea", "NoMatchingParent"}, + "namespace not selected": {func(in *GatewayInput) { in.Namespaces[0].Labels = nil }, "gitea", "NotAllowedByListeners"}, + "hostname outside the listener's": {func(in *GatewayInput) { in.HTTPRoutes[0].Spec.Hostnames = []string{"git.example.net"} }, "gitea", "NoMatchingListenerHostname"}, + "unsupported filter": {func(in *GatewayInput) { + in.HTTPRoutes[0].Spec.Rules[0].Filters = []HTTPRouteFilter{{Type: "RequestMirror"}} + }, "gitea", "UnsupportedValue"}, + "regular expression path": {func(in *GatewayInput) { + in.HTTPRoutes[0].Spec.Rules[0].Matches = []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: ptr("RegularExpression"), Value: ptr("/.*")}}} + }, "gitea", "UnsupportedValue"}, + "wrong port": {func(in *GatewayInput) { in.TCPRoutes[0].Spec.ParentRefs[0].Port = ptr(int32(2222)) }, "gitea-ssh", "NoMatchingParent"}, + } + for name, c := range cases { + in := giteaInput(t) + c.mutate(&in) + d := ComputeGateways("cp", gwNodes(), in) + found := false + for _, r := range d.Routes { + if r.Name != c.route { + continue + } + found = true + a := condOf(r.Status.Parents[0].Conditions, condAccepted) + if a.Status != metav1.ConditionFalse || a.Reason != c.reason { + t.Errorf("%s: accepted %+v", name, a) + } + } + if !found { + t.Errorf("%s: no status for route %s", name, c.route) + } + h := ComputeGateways("w15", gwNodes(), in) + if c.route == "gitea" { + for _, s := range h.Serve { + if s.Name == "https-gitea" && len(s.Routes) != 0 { + t.Errorf("%s: a route that is not accepted was attached", name) + } + } + } + } +} + +func TestBackendNotFoundBreaksTheRule(t *testing.T) { + in := giteaInput(t) + in.HTTPRoutes[0].Spec.Rules[0].BackendRefs[0].Name = "nope" + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + if r.Name == "gitea" { + if c := condOf(r.Status.Parents[0].Conditions, condResolvedRefs); c.Status != metav1.ConditionFalse || c.Reason != "BackendNotFound" { + t.Fatalf("resolvedRefs: %+v", c) + } + if c := condOf(r.Status.Parents[0].Conditions, condAccepted); c.Status != metav1.ConditionTrue { + t.Fatalf("a route with an unresolved backend is still accepted: %+v", c) + } + } + } + h := ComputeGateways("w15", gwNodes(), in) + for _, s := range h.Serve { + if s.Name == "https-gitea" && (len(s.Routes) != 1 || !s.Routes[0].Rules[0].Broken) { + t.Fatalf("the rule must be served broken (500): %+v", s.Routes) + } + } +} + +func TestTCPListenerForwardsTheOldestRoute(t *testing.T) { + in := giteaInput(t) + later := in.TCPRoutes[0].DeepCopyObject().(*TCPRoute) + later.Name = "other-ssh" + later.CreationTimestamp = metav1.NewTime(later.CreationTimestamp.Add(time.Minute)) + later.Spec.Rules[0].BackendRefs[0].Name = "other" + in.TCPRoutes = append(in.TCPRoutes, later) + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + a := condOf(r.Status.Parents[0].Conditions, condAccepted) + switch r.Name { + case "gitea-ssh": + if a.Status != metav1.ConditionTrue { + t.Fatalf("oldest route: %+v", a) + } + case "other-ssh": + if a.Status != metav1.ConditionFalse || !strings.Contains(a.Message, "gitea/gitea-ssh") { + t.Fatalf("second route: %+v", a) + } + } + } +} + +func TestOtherControllersParentsKept(t *testing.T) { + in := giteaInput(t) + theirs := RouteParentStatus{ParentRef: ParentReference{Name: "elsewhere"}, ControllerName: "example.com/other", + Conditions: []metav1.Condition{{Type: condAccepted, Status: metav1.ConditionTrue, Reason: "Accepted", LastTransitionTime: gwNow}}} + in.HTTPRoutes[0].Status.Parents = []RouteParentStatus{theirs} + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + if r.Name == "gitea" { + if len(r.Status.Parents) != 2 || r.Status.Parents[0].ControllerName != "example.com/other" { + t.Fatalf("another controller's entry must be kept: %+v", r.Status.Parents) + } + } + } +} + +func TestOtherClassesIgnored(t *testing.T) { + in := giteaInput(t) + in.Classes[0].Spec.ControllerName = "example.com/other" + for _, self := range []string{"cp", "w15"} { + d := ComputeGateways(self, gwNodes(), in) + if len(d.Serve)+len(d.Forwards)+len(d.Gateways)+len(d.Routes)+len(d.Classes) != 0 { + t.Fatalf("%s acted on another controller's class: %+v", self, d) + } + } +} + +func TestNarrowHostnames(t *testing.T) { + cases := []struct { + listener string + route []string + want []string + ok bool + }{ + {"", nil, nil, true}, + {"git.levk.dev", nil, []string{"git.levk.dev"}, true}, + {"", []string{"a.example.com"}, []string{"a.example.com"}, true}, + {"test.example.com", []string{"*.example.com"}, []string{"test.example.com"}, true}, + {"*.example.com", []string{"test.example.com", "test.example.net", "foo.test.example.com"}, []string{"test.example.com", "foo.test.example.com"}, true}, + {"*.example.com", []string{"example.com"}, nil, false}, + {"*.example.com", []string{"*.a.example.com"}, []string{"*.a.example.com"}, true}, + {"*.a.example.com", []string{"*.example.com"}, []string{"*.a.example.com"}, true}, + {"git.levk.dev", []string{"git.example.net"}, nil, false}, + } + for _, c := range cases { + got, ok := narrowHostnames(c.listener, c.route) + if ok != c.ok || strings.Join(got, ",") != strings.Join(c.want, ",") { + t.Errorf("listener %q route %v: got %v %v, want %v %v", c.listener, c.route, got, ok, c.want, c.ok) + } + } +} + +func TestMergeGateways(t *testing.T) { + in := giteaInput(t) + nodes := gwNodes() + // on the holder: Gitea's SSH forwarded, its HTTP(S) ports let in + d := MergeGateways(Compute("w15", nodes, nil, nil), ComputeGateways("w15", nodes, in), "w15", nodes) + if len(d.Forwards) != 1 || len(d.Accept) != 2 || len(d.Holders) != 0 { + t.Fatalf("holder: forwards %+v accept %+v holders %+v", d.Forwards, d.Accept, d.Holders) + } + r := Render(d, EgressDesired{}, "w15", &net.IPNet{IP: net.ParseIP("10.32.0.0"), Mask: net.CIDRMask(12, 32)}, nil) + for _, want := range []string{ + "-A REWEAVE-LB-IN -d 86.107.168.15/32 -p tcp -m tcp --dport 443", + "-A REWEAVE-LB-IN -d 86.107.168.15/32 -p tcp -m tcp --dport 80", + } { + if !strings.Contains(r.Filter, want) { + t.Errorf("filter lacks %q:\n%s", want, r.Filter) + } + } + if !strings.Contains(r.Nat, "-A REWEAVE-LB -d 86.107.168.15/32 -p tcp -m tcp --dport 22") || !strings.Contains(r.Nat, "--to-destination 10.44.0.12:22") { + t.Errorf("nat lacks the SSH forward:\n%s", r.Nat) + } + // elsewhere: w15 is a holder whose replies come back to it + e := MergeGateways(Compute("cp", nodes, nil, nil), ComputeGateways("cp", nodes, in), "cp", nodes) + if len(e.Forwards) != 0 || len(e.Accept) != 0 || len(e.Holders) != 1 || e.Holders[0].Node.Name != "w15" { + t.Fatalf("cp: forwards %+v accept %+v holders %+v", e.Forwards, e.Accept, e.Holders) + } +} + +func TestUnboundListenerReported(t *testing.T) { + in := giteaInput(t) + in.Unbound = map[string]string{"gateway-system/shared/https-gitea": "address already in use"} + d := ComputeGateways("w15", gwNodes(), in) + for _, l := range d.Gateways[0].Status.Listeners { + if l.Name != "https-gitea" { + continue + } + if a := condOf(l.Conditions, condAccepted); a.Status != metav1.ConditionFalse || a.Reason != "PortUnavailable" { + t.Fatalf("accepted: %+v", a) + } + if p := condOf(l.Conditions, condProgrammed); p.Status != metav1.ConditionFalse { + t.Fatalf("programmed: %+v", p) + } + } + served := false + for _, s := range d.Serve { + served = served || s.Name == "https-gitea" + } + if !served { + t.Fatal("an unbound listener is still tried on every pass") + } +} diff --git a/lb/model.go b/lb/model.go index df208711..c25c6fe0 100644 --- a/lb/model.go +++ b/lb/model.go @@ -136,9 +136,19 @@ type Desired struct { Forwards []Forward Holders []Holder Status []Status + // Accept are the address:ports this node serves itself (Gateways' HTTP and HTTPS listeners), let in ahead of + // the host firewall + Accept []Accept Problems []string // what an operator should hear about (logged and evented by the controller) } +// Accept is a TCP address:port this node serves itself. +type Accept struct { + Address net.IP + Port int32 + Comment string +} + // eligible reports whether the Service asks for this agent. func eligible(s *coreapi.Service) bool { return s.Spec.Type == coreapi.ServiceTypeLoadBalancer && s.Spec.LoadBalancerClass != nil && diff --git a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml index e1de7041..84faf1ce 100644 --- a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml +++ b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml @@ -154,6 +154,28 @@ items: - egressservices/status verbs: - patch + # Gateways of controller reweave.levk.dev/gateway (the Gateway API CRDs are installed separately) + - apiGroups: + - gateway.networking.k8s.io + resources: + - gatewayclasses + - gateways + - httproutes + - tcproutes + - referencegrants + verbs: + - get + - list + - watch + - apiGroups: + - gateway.networking.k8s.io + resources: + - gatewayclasses/status + - gateways/status + - httproutes/status + - tcproutes/status + verbs: + - patch - apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: @@ -206,6 +228,42 @@ items: - kind: ServiceAccount name: weave-net namespace: kube-system + # Gateways' certificates: the agent reads Secrets in gateway-system only (weave-npc --gateway-namespaces) + - apiVersion: v1 + kind: Namespace + metadata: + name: gateway-system + - apiVersion: rbac.authorization.k8s.io/v1 + kind: Role + metadata: + name: weave-net-gateway-certificates + namespace: gateway-system + labels: + name: weave-net + rules: + - apiGroups: + - '' + resources: + - secrets + verbs: + - get + - list + - watch + - apiVersion: rbac.authorization.k8s.io/v1 + kind: RoleBinding + metadata: + name: weave-net-gateway-certificates + namespace: gateway-system + labels: + name: weave-net + roleRef: + kind: Role + name: weave-net-gateway-certificates + apiGroup: rbac.authorization.k8s.io + subjects: + - kind: ServiceAccount + name: weave-net + namespace: kube-system - apiVersion: apps/v1 kind: DaemonSet metadata: diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index a7b2b369..48a0bd1f 100644 --- a/prog/weave-npc/main.go +++ b/prog/weave-npc/main.go @@ -6,6 +6,7 @@ import ( "os" "os/signal" goruntime "runtime" + "strings" "syscall" "github.com/coreos/go-iptables/iptables" @@ -30,14 +31,15 @@ import ( ) var ( - version = "unreleased" - metricsAddr string - logLevel string - allowMcast bool - nodeName string - maxList int - bridgePortName string - serveLB bool + version = "unreleased" + metricsAddr string + logLevel string + allowMcast bool + nodeName string + maxList int + bridgePortName string + serveLB bool + gatewayNamespaces string ) func handleError(err error) { @@ -344,7 +346,14 @@ func root(cmd *cobra.Command, args []string) { // EgressServices: those Services' pods leave the cluster from that address (or not at all) egress, err := lb.NewEgressClient(config) handleFatal(err) - go lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)).Run(context.Background()) + ctl := lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)) + if gatewayNamespaces != "" { + // and Gateways of class controller lb.GatewayControllerName, from the node holding their address + gw, err := lb.NewGatewayClient(config) + handleFatal(err) + ctl = ctl.WithGateway(gw, strings.Split(gatewayNamespaces, ",")) + } + go ctl.Run(context.Background()) } signals := make(chan os.Signal, 1) @@ -365,6 +374,8 @@ func main() { rootCmd.PersistentFlags().IntVar(&maxList, "max-list-size", 1024, "maximum size of ipset list (for namespaces)") rootCmd.PersistentFlags().StringVar(&bridgePortName, "bridge-port-name", "vethwe-bridge", "name of the brige port on which packets are received and sent") rootCmd.PersistentFlags().BoolVar(&serveLB, "lb", true, "serve LoadBalancer Services of class "+lb.Class+" from addresses nodes hold") + rootCmd.PersistentFlags().StringVar(&gatewayNamespaces, "gateway-namespaces", "gateway-system", + "serve Gateways of controller "+lb.GatewayControllerName+", with certificates from Secrets in these namespaces (comma-separated; empty: no Gateways)") handleFatal(rootCmd.Execute()) }