images / images (push) Successful in 6m32s
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
1010 lines
34 KiB
Go
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
|
|
}
|