// Package lb gives LoadBalancer Services an address that a node holds, and // forwards that address's traffic to the Service's endpoints with the // client's address intact. // // A Service opts in with spec.loadBalancerClass = Class and names its // address in spec.loadBalancerIP. The node that holds the address (has it on // one of its interfaces, as published in its AnnAddresses annotation) is the // address's holder: it alone writes the Service's status (ipMode Proxy, so // kube-proxy installs no rules for the address) and DNATs the address's // ports to ready endpoints, without SNAT. Every other node marks the // connections that the holder forwards to its pods, and routes their replies // back to the holder over weave, so that the holder's conntrack can undo the // DNAT and the reply leaves from the address the client connected to. // // Compute is the whole policy as a pure function of what the API shows; the // datapath and the controller only carry it out. package lb import ( "fmt" "hash/fnv" "net" "sort" "strings" coreapi "k8s.io/api/core/v1" discovery "k8s.io/api/discovery/v1" ) const ( // Class is the spec.loadBalancerClass a Service sets to be served by this agent. Class = "reweave.levk.dev/node-address" // AnnAddresses lists the IPv4 addresses a node holds on its host interfaces (comma-separated). AnnAddresses = "reweave.levk.dev/addresses" // AnnWeave is "," of the node's weave bridge: where replies to its forwards go. AnnWeave = "reweave.levk.dev/weave" serviceNameLabel = "kubernetes.io/service-name" ) // Node is what one node publishes about itself. type Node struct { Name string Addresses []net.IP WeaveIP net.IP WeaveMAC net.HardwareAddr } // Holds reports whether the node has ip on one of its host interfaces. func (n Node) Holds(ip net.IP) bool { for _, a := range n.Addresses { if a.Equal(ip) { return true } } return false } // ParseNode reads a Node's annotations. A node that has published nothing // (yet) holds nothing and cannot take replies. func ParseNode(n *coreapi.Node) (Node, error) { out := Node{Name: n.Name} if s := n.Annotations[AnnAddresses]; s != "" { for _, f := range strings.Split(s, ",") { ip := net.ParseIP(strings.TrimSpace(f)).To4() if ip == nil { return out, fmt.Errorf("node %s: %s: %q is not an IPv4 address", n.Name, AnnAddresses, f) } out.Addresses = append(out.Addresses, ip) } } if s := n.Annotations[AnnWeave]; s != "" { f := strings.Split(s, ",") if len(f) != 2 { return out, fmt.Errorf("node %s: %s: want \",\", got %q", n.Name, AnnWeave, s) } ip := net.ParseIP(f[0]).To4() mac, err := net.ParseMAC(f[1]) if ip == nil || err != nil { return out, fmt.Errorf("node %s: %s: want \",\", got %q", n.Name, AnnWeave, s) } out.WeaveIP, out.WeaveMAC = ip, mac } return out, nil } // FormatAddresses is the AnnAddresses value for a set of addresses (sorted, so it only changes when the set does). func FormatAddresses(ips []net.IP) string { s := make([]string, 0, len(ips)) for _, ip := range ips { s = append(s, ip.String()) } sort.Strings(s) return strings.Join(s, ",") } // Endpoint is one ready backend of one Service port. type Endpoint struct { IP net.IP Port int32 } func (e Endpoint) String() string { return fmt.Sprintf("%s:%d", e.IP, e.Port) } // Port is one Service port at the held address. type Port struct { Protocol coreapi.Protocol Port int32 Endpoints []Endpoint } // Forward is a Service whose address this node holds. type Forward struct { Namespace, Name string Address net.IP Ports []Port } // Status is a write to a Service's status.loadBalancer: Address set, or cleared when nil. type Status struct { Namespace, Name string Address net.IP } // Holder is another node that forwards held addresses' connections to pods; Mark identifies its connections here. type Holder struct { Node Node Mark uint32 // 1..MaxMarks, in the connmark's top byte } // MaxMarks is how many holders a node can tell apart (the top byte of the mark, 0 meaning none). const MaxMarks = 250 // Desired is what one node must do. type Desired struct { Forwards []Forward Holders []Holder Status []Status Problems []string // what an operator should hear about (logged and evented by the controller) } // 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 && *s.Spec.LoadBalancerClass == Class } // shown is the address status.loadBalancer currently shows (nil if none or not a single IP). func shown(s *coreapi.Service) net.IP { if len(s.Status.LoadBalancer.Ingress) != 1 { return nil } return net.ParseIP(s.Status.LoadBalancer.Ingress[0].IP).To4() } func hasStatus(s *coreapi.Service) bool { return len(s.Status.LoadBalancer.Ingress) != 0 } // Compute decides what node `self` does, given every node, every Service and every EndpointSlice. func Compute(self string, nodes []Node, services []*coreapi.Service, slices []*discovery.EndpointSlice) Desired { var d Desired sortedNodes := append([]Node(nil), nodes...) sort.Slice(sortedNodes, func(i, j int) bool { return sortedNodes[i].Name < sortedNodes[j].Name }) holdersOf := func(ip net.IP) []Node { var hs []Node for _, n := range sortedNodes { if ip != nil && n.Holds(ip) { hs = append(hs, n) } } return hs } // who may write a status that shows `ip`: its single holder, or (nobody holds it) the first node by name writerFor := func(ip net.IP) string { if hs := holdersOf(ip); len(hs) == 1 { return hs[0].Name } if len(sortedNodes) > 0 { return sortedNodes[0].Name } return "" } holding := map[string]bool{} // nodes that forward some address svcs := append([]*coreapi.Service(nil), services...) sort.Slice(svcs, func(i, j int) bool { return key(svcs[i]) < key(svcs[j]) }) for _, s := range svcs { if !eligible(s) { // loadBalancerClass is immutable and only allowed on type LoadBalancer, so a // Service never leaves the class while keeping a status we wrote continue } want := net.ParseIP(s.Spec.LoadBalancerIP).To4() var holder *Node switch hs := holdersOf(want); { case want == nil: d.Problems = append(d.Problems, fmt.Sprintf("%s: spec.loadBalancerIP must name an IPv4 address a node holds (got %q)", key(s), s.Spec.LoadBalancerIP)) case len(hs) == 0: d.Problems = append(d.Problems, fmt.Sprintf("%s: no node holds %s", key(s), want)) case len(hs) > 1: names := make([]string, len(hs)) for i, h := range hs { names[i] = h.Name } d.Problems = append(d.Problems, fmt.Sprintf("%s: %s is held by several nodes (%s); serving it from none", key(s), want, strings.Join(names, ", "))) default: holder = &hs[0] } if holder == nil { // fail closed: nobody forwards, and the status must not claim an address if hasStatus(s) && writerFor(shown(s)) == self { d.Status = append(d.Status, Status{Namespace: s.Namespace, Name: s.Name}) } continue } holding[holder.Name] = true if holder.Name != self { continue } if cur := shown(s); cur == nil || !cur.Equal(want) { d.Status = append(d.Status, Status{Namespace: s.Namespace, Name: s.Name, Address: want}) } d.Forwards = append(d.Forwards, Forward{Namespace: s.Namespace, Name: s.Name, Address: want, Ports: ports(s, slices)}) } var others []Node for _, n := range sortedNodes { if holding[n.Name] && n.Name != self { 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) return d } // assignMarks gives each holder a mark from its name, probing past collisions in name order: stable while the set is. func assignMarks(nodes []Node) []Holder { taken := map[uint32]bool{} out := make([]Holder, 0, len(nodes)) for _, n := range nodes { if len(out) == MaxMarks { break } h := fnv.New32a() h.Write([]byte(n.Name)) m := h.Sum32()%MaxMarks + 1 for taken[m] { m = m%MaxMarks + 1 } taken[m] = true out = append(out, Holder{Node: n, Mark: m}) } return out } func key(s *coreapi.Service) string { return s.Namespace + "/" + s.Name } // ports pairs each Service port with its ready endpoints. func ports(s *coreapi.Service, slices []*discovery.EndpointSlice) []Port { var out []Port for _, sp := range s.Spec.Ports { proto := sp.Protocol if proto == "" { proto = coreapi.ProtocolTCP } seen := map[string]bool{} var eps []Endpoint for _, sl := range slices { if sl.Namespace != s.Namespace || sl.Labels[serviceNameLabel] != s.Name || sl.AddressType != discovery.AddressTypeIPv4 { continue } for _, p := range sl.Ports { pname, pproto := "", coreapi.ProtocolTCP if p.Name != nil { pname = *p.Name } if p.Protocol != nil { pproto = *p.Protocol } if pname != sp.Name || pproto != proto || p.Port == nil { continue } for _, ep := range sl.Endpoints { if ep.Conditions.Ready != nil && !*ep.Conditions.Ready { continue } for _, a := range ep.Addresses { ip := net.ParseIP(a).To4() if ip == nil { continue } e := Endpoint{IP: ip, Port: *p.Port} if !seen[e.String()] { seen[e.String()] = true eps = append(eps, e) } } } } } sort.Slice(eps, func(i, j int) bool { return eps[i].String() < eps[j].String() }) out = append(out, Port{Protocol: proto, Port: sp.Port, Endpoints: eps}) } return out }