Files
weave/lb/gateway.go
levaliciousandClaude Opus 5.5 7405b51e53
images / images (push) Successful in 6m32s
lb: serve Gateways (Gateway API v1) from the node holding their address
Gateways of a GatewayClass whose controllerName is reweave.levk.dev/gateway
are served like LoadBalancer Services: a Gateway names one IPv4 address in
spec.addresses, and the node holding that address serves all of it and
writes its status.

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

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

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

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

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ
2026-10-13 23:51:44 -04:00

1010 lines
34 KiB
Go

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
}