diff --git a/lb/addresses.go b/lb/addresses.go new file mode 100644 index 00000000..1a2fa512 --- /dev/null +++ b/lb/addresses.go @@ -0,0 +1,87 @@ +package lb + +import ( + "fmt" + "net" + "sort" + "strings" + + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +// Local is what this node publishes about itself, read from its interfaces. +type Local struct { + Addresses []net.IP + WeaveIP net.IP + WeaveMAC net.HardwareAddr + PodNet *net.IPNet // the weave allocation range: the bridge address's network +} + +// overlay- and container-side interfaces: their addresses are never held ones +var overlayPrefixes = []string{"weave", "datapath", "vxlan-", "veth", "docker", "cni"} + +func overlay(name string) bool { + for _, p := range overlayPrefixes { + if strings.HasPrefix(name, p) { + return true + } + } + return false +} + +// ReadLocal lists the global IPv4 addresses on host interfaces, and the weave bridge's address and MAC. +func ReadLocal() (Local, error) { + var l Local + links, err := netlink.LinkList() + if err != nil { + return l, err + } + for _, link := range links { + name := link.Attrs().Name + addrs, err := netlink.AddrList(link, unix.AF_INET) + if err != nil { + return l, fmt.Errorf("%s: %v", name, err) + } + if name == bridge { + for _, a := range addrs { + if a.IPNet != nil && a.IP.To4() != nil { + l.WeaveIP = a.IP.To4() + l.WeaveMAC = link.Attrs().HardwareAddr + l.PodNet = &net.IPNet{IP: a.IP.Mask(a.Mask), Mask: a.Mask} + break + } + } + continue + } + if overlay(name) || link.Attrs().Flags&net.FlagLoopback != 0 { + continue + } + for _, a := range addrs { + if a.Scope == unix.RT_SCOPE_UNIVERSE && a.IP.To4() != nil && a.IP.IsGlobalUnicast() { + l.Addresses = append(l.Addresses, a.IP.To4()) + } + } + } + // an address inside the pod network is never a host one (whatever interface it is on) + if l.PodNet != nil { + kept := l.Addresses[:0] + for _, ip := range l.Addresses { + if !l.PodNet.Contains(ip) { + kept = append(kept, ip) + } + } + l.Addresses = kept + } + sort.Slice(l.Addresses, func(i, j int) bool { return l.Addresses[i].String() < l.Addresses[j].String() }) + return l, nil +} + +// Annotations is what this node's Node object should carry for l. +func (l Local) Annotations() map[string]string { + a := map[string]string{AnnAddresses: FormatAddresses(l.Addresses)} + if l.WeaveIP != nil && l.WeaveMAC != nil { + a[AnnWeave] = l.WeaveIP.String() + "," + l.WeaveMAC.String() + } + return a +} diff --git a/lb/controller.go b/lb/controller.go new file mode 100644 index 00000000..9589b8ba --- /dev/null +++ b/lb/controller.go @@ -0,0 +1,221 @@ +package lb + +import ( + "context" + "encoding/json" + "fmt" + "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/fields" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" + + "github.com/rajch/weave/common" +) + +// Controller watches Services, EndpointSlices and Nodes and keeps this node's part of the picture true. +type Controller struct { + self string + client kubernetes.Interface + dp *Datapath + // readLocal is the node's own interfaces (swappable in tests) + readLocal func() (Local, error) + + services, slices, nodes cache.Store + kick chan struct{} + lastProblems map[string]bool + // Services whose status this node wrote since it started (so a restart rewrites ipMode once) + written map[types.UID]string +} + +// Resync is how often the node re-reads its interfaces and re-applies everything even without API events. +const Resync = 30 * time.Second + +// NewController runs nothing yet; Run does. +func NewController(self string, client kubernetes.Interface, dp *Datapath) *Controller { + return &Controller{self: self, client: client, dp: dp, readLocal: ReadLocal, + kick: make(chan struct{}, 1), lastProblems: map[string]bool{}, written: map[types.UID]string{}} +} + +func (c *Controller) poke() { + select { + case c.kick <- struct{}{}: + default: + } +} + +// Run blocks until ctx ends. +func (c *Controller) Run(ctx context.Context) { + var sc, ec, nc cache.Controller + c.services, sc = newInformer(c, c.client.CoreV1().RESTClient(), "services", &coreapi.Service{}) + c.slices, ec = newInformer(c, c.client.DiscoveryV1().RESTClient(), "endpointslices", &discovery.EndpointSlice{}) + c.nodes, nc = newInformer(c, c.client.CoreV1().RESTClient(), "nodes", &coreapi.Node{}) + for _, ctl := range []cache.Controller{sc, ec, nc} { + go ctl.Run(ctx.Done()) + } + if !cache.WaitForCacheSync(ctx.Done(), sc.HasSynced, ec.HasSynced, nc.HasSynced) { + return + } + tick := time.NewTicker(Resync) + defer tick.Stop() + c.poke() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + case <-c.kick: + // let a burst of events settle into one pass + time.Sleep(200 * time.Millisecond) + select { + case <-c.kick: + default: + } + } + if err := c.sync(ctx); err != nil { + common.Log.Errorf("lb: %v", err) + } + } +} + +func newInformer(c *Controller, getter cache.Getter, resource string, obj runtime.Object) (cache.Store, cache.Controller) { + lw := cache.NewListWatchFromClient(getter, resource, "", fields.Everything()) + return cache.NewInformer(lw, obj, 0, cache.ResourceEventHandlerFuncs{ + AddFunc: func(interface{}) { c.poke() }, + UpdateFunc: func(interface{}, interface{}) { c.poke() }, + DeleteFunc: func(interface{}) { c.poke() }, + }) +} + +// sync is one pass: publish this node, compute, apply, write statuses. +func (c *Controller) sync(ctx context.Context) error { + local, err := c.readLocal() + if err != nil { + return fmt.Errorf("reading interfaces: %v", err) + } + if err := c.publish(ctx, local); err != nil { + return err + } + if local.PodNet == nil { + return fmt.Errorf("weave bridge has no address yet; not forwarding") + } + + var nodes []Node + var problems []string + for _, o := range c.nodes.List() { + n, err := ParseNode(o.(*coreapi.Node)) + if err != nil { + problems = append(problems, err.Error()) + } + nodes = append(nodes, n) + } + var svcs []*coreapi.Service + for _, o := range c.services.List() { + svcs = append(svcs, o.(*coreapi.Service)) + } + var slices []*discovery.EndpointSlice + for _, o := range c.slices.List() { + slices = append(slices, o.(*discovery.EndpointSlice)) + } + d := Compute(c.self, nodes, svcs, slices) + // 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) { + d.Status = append(d.Status, Status{Namespace: f.Namespace, Name: f.Name, Address: f.Address}) + } + } + current := map[string]bool{} + for _, p := range append(problems, d.Problems...) { + current[p] = true + if !c.lastProblems[p] { + common.Log.Warnf("lb: %s", p) + } + } + c.lastProblems = current + + if err := c.dp.Apply(d, local.PodNet); err != nil { + return fmt.Errorf("applying datapath: %v", err) + } + for _, st := range d.Status { + if err := c.writeStatus(ctx, st); err != nil { + return err + } + } + return nil +} + +func hasStatusFor(d Desired, f Forward) bool { + for _, s := range d.Status { + if s.Namespace == f.Namespace && s.Name == f.Name { + return true + } + } + return false +} + +func (c *Controller) service(ns, name string) *coreapi.Service { + o, ok, err := c.services.GetByKey(ns + "/" + name) + if err != nil || !ok { + return nil + } + return o.(*coreapi.Service) +} + +// publish keeps this node's annotations equal to what its interfaces say. +func (c *Controller) publish(ctx context.Context, local Local) error { + want := local.Annotations() + o, ok, err := c.nodes.GetByKey(c.self) + if err != nil { + return err + } + if ok { + have := o.(*coreapi.Node).Annotations + same := true + for k, v := range want { + if have[k] != v { + same = false + } + } + if same { + return nil + } + } + patch, err := json.Marshal(map[string]interface{}{"metadata": map[string]interface{}{"annotations": want}}) + if err != nil { + return err + } + if _, err := c.client.CoreV1().Nodes().Patch(ctx, c.self, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + return fmt.Errorf("publishing node annotations: %v", err) + } + common.Log.Infof("lb: node %s holds [%s]", c.self, want[AnnAddresses]) + return nil +} + +// writeStatus sets (ipMode Proxy: kube-proxy leaves the address to us) or clears a Service's load-balancer status. +func (c *Controller) writeStatus(ctx context.Context, st Status) error { + ingress := []map[string]string{} + if st.Address != nil { + ingress = append(ingress, map[string]string{"ip": st.Address.String(), "ipMode": "Proxy"}) + } + patch, err := json.Marshal(map[string]interface{}{"status": map[string]interface{}{"loadBalancer": map[string]interface{}{"ingress": ingress}}}) + if err != nil { + return err + } + s, err := c.client.CoreV1().Services(st.Namespace).Patch(ctx, st.Name, types.MergePatchType, patch, metav1.PatchOptions{}, "status") + if err != nil { + return fmt.Errorf("status of %s/%s: %v", st.Namespace, st.Name, err) + } + if st.Address != nil { + c.written[s.UID] = st.Address.String() + common.Log.Infof("lb: %s/%s at %s", st.Namespace, st.Name, st.Address) + } else { + delete(c.written, s.UID) + common.Log.Infof("lb: %s/%s has no address", st.Namespace, st.Name) + } + return nil +} diff --git a/lb/datapath.go b/lb/datapath.go new file mode 100644 index 00000000..3fb06b2a --- /dev/null +++ b/lb/datapath.go @@ -0,0 +1,254 @@ +package lb + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "fmt" + "net" + "os/exec" + "strings" + + "github.com/coreos/go-iptables/iptables" + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +const ( + // ForwardChain (nat, from PREROUTING) DNATs held addresses' ports to endpoints. + ForwardChain = "REWEAVE-LB" + // MarkChain (mangle, from PREROUTING) marks holders' connections and steers their replies. + MarkChain = "REWEAVE-LB-MARK" + // servicePrefix names the per-port chains ForwardChain jumps to. + servicePrefix = "REWEAVE-LBS-" + + markShift = 24 + markMask = uint32(0xff) << markShift + // RulePriority is where the reply-steering rules sit: before the main table (32766). + RulePriority = 9000 + // TableBase + mark is the routing table that sends replies to that mark's holder. + TableBase = 52000 + + bridge = "weave" +) + +func markValue(m uint32) uint32 { return m << markShift } + +// portChain is the per-(service, protocol, port) chain name: stable, short enough for iptables (28 chars). +func portChain(f Forward, p Port) string { + h := sha256.Sum256([]byte(fmt.Sprintf("%s/%s/%s/%d", f.Namespace, f.Name, p.Protocol, p.Port))) + return servicePrefix + strings.ToUpper(hex.EncodeToString(h[:])[:12]) +} + +// Rules is the full content of the chains this package owns, in iptables-restore form. +type Rules struct { + Nat, Mangle string +} + +// Render turns the desired state into iptables-restore input for the nat and mangle tables. +// podNet is the weave allocation range (pods and bridges); bridge is the weave bridge name. +// Each payload declares every chain it owns, so a restore with --noflush replaces exactly those. +func Render(d Desired, podNet *net.IPNet, stale []string) Rules { + var nat bytes.Buffer + nat.WriteString("*nat\n") + fmt.Fprintf(&nat, ":%s - [0:0]\n", ForwardChain) + var body bytes.Buffer + for _, f := range d.Forwards { + for _, p := range f.Ports { + c := portChain(f, p) + fmt.Fprintf(&nat, ":%s - [0:0]\n", c) + proto := strings.ToLower(string(p.Protocol)) + fmt.Fprintf(&body, "-A %s -d %s/32 -p %s -m %s --dport %d -m comment --comment %q -j %s\n", + ForwardChain, f.Address, proto, proto, p.Port, f.Namespace+"/"+f.Name, c) + n := len(p.Endpoints) + for i, e := range p.Endpoints { + if i < n-1 { + // the i-th endpoint takes 1/(n-i) of what the earlier ones left: uniform overall + fmt.Fprintf(&body, "-A %s -m statistic --mode random --probability %.10f -j DNAT --to-destination %s\n", + c, 1/float64(n-i), e) + } else { + fmt.Fprintf(&body, "-A %s -j DNAT --to-destination %s\n", c, e) + } + } + } + } + for _, c := range stale { + // chains of ports no longer forwarded: emptied here, deleted after the restore + fmt.Fprintf(&nat, ":%s - [0:0]\n", c) + } + nat.Write(body.Bytes()) + nat.WriteString("COMMIT\n") + + var mangle bytes.Buffer + mangle.WriteString("*mangle\n") + fmt.Fprintf(&mangle, ":%s - [0:0]\n", MarkChain) + if len(d.Holders) > 0 { + // replies of a holder's connection carry its mark: the policy rule routes them back to it + fmt.Fprintf(&mangle, "-A %s -m conntrack --ctdir REPLY -m connmark ! --mark 0x0/0x%x -j CONNMARK --restore-mark --nfmask 0x%x --ctmask 0x%x\n", + MarkChain, markMask, markMask, markMask) + for _, h := range d.Holders { + // a new connection to a pod here, arriving from the holder's bridge, from outside the pod network: + // the holder DNATed it and must see its replies + fmt.Fprintf(&mangle, "-A %s -i %s -m mac --mac-source %s ! -s %s -d %s -m conntrack --ctstate NEW -m comment --comment %q -j CONNMARK --set-xmark 0x%x/0x%x\n", + MarkChain, bridge, h.Node.WeaveMAC, podNet, podNet, "holder "+h.Node.Name, markValue(h.Mark), markMask) + } + } + mangle.WriteString("COMMIT\n") + return Rules{Nat: nat.String(), Mangle: mangle.String()} +} + +// Datapath applies Desired to this node. +type Datapath struct { + ipt *iptables.IPTables + // restore runs iptables-restore --noflush with the given input (swappable in tests) + restore func(input string) error +} + +// NewDatapath uses the host's iptables (whichever backend the image's launch script selected). +func NewDatapath(ipt *iptables.IPTables) *Datapath { + return &Datapath{ipt: ipt, restore: func(input string) error { + cmd := exec.Command("iptables-restore", "--noflush", "-w", "5") + cmd.Stdin = strings.NewReader(input) + if out, err := cmd.CombinedOutput(); err != nil { + return fmt.Errorf("iptables-restore: %v: %s", err, out) + } + return nil + }} +} + +// Apply makes the node's iptables chains, policy rules and routes exactly what d says. +func (dp *Datapath) Apply(d Desired, podNet *net.IPNet) error { + if err := dp.hooks(); err != nil { + return err + } + existing, err := dp.ipt.ListChains("nat") + if err != nil { + return err + } + want := map[string]bool{} + for _, f := range d.Forwards { + for _, p := range f.Ports { + want[portChain(f, p)] = true + } + } + var stale []string + for _, c := range existing { + if strings.HasPrefix(c, servicePrefix) && !want[c] { + stale = append(stale, c) + } + } + r := Render(d, podNet, stale) + if err := dp.restore(r.Nat); err != nil { + return err + } + if err := dp.restore(r.Mangle); err != nil { + return err + } + for _, c := range stale { + if err := dp.ipt.DeleteChain("nat", c); err != nil { + return fmt.Errorf("deleting stale chain %s: %v", c, err) + } + } + return dp.routes(d.Holders) +} + +// hooks makes sure PREROUTING jumps to our chains first (kube-proxy's KUBE-SERVICES must not see held addresses first). +func (dp *Datapath) hooks() error { + for _, h := range []struct{ table, chain string }{{"nat", ForwardChain}, {"mangle", MarkChain}} { + ok, err := dp.ipt.ChainExists(h.table, h.chain) + if err != nil { + return err + } + if !ok { + if err := dp.ipt.NewChain(h.table, h.chain); err != nil { + return err + } + } + rule := []string{"-m", "comment", "--comment", "reweave lb", "-j", h.chain} + rules, err := dp.ipt.List(h.table, "PREROUTING") + if err != nil { + return err + } + // rules[0] is the policy line; our jump must be rules[1] + if len(rules) > 1 && strings.HasSuffix(rules[1], "-j "+h.chain) { + continue + } + if err := dp.ipt.DeleteIfExists(h.table, "PREROUTING", rule...); err != nil { + return err + } + if err := dp.ipt.Insert(h.table, "PREROUTING", 1, rule...); err != nil { + return err + } + } + return nil +} + +// routes keeps one policy rule and one table per holder: marked replies go to the holder's bridge. +func (dp *Datapath) routes(holders []Holder) error { + link, err := netlink.LinkByName(bridge) + if err != nil { + if len(holders) == 0 { + return nil + } + return fmt.Errorf("weave bridge: %v", err) + } + want := map[int]Holder{} + for _, h := range holders { + want[TableBase+int(h.Mark)] = h + } + rules, err := netlink.RuleList(unix.AF_INET) + if err != nil { + return err + } + have := map[int]bool{} + for _, r := range rules { + if r.Priority != RulePriority || r.Table < TableBase || r.Table > TableBase+MaxMarks { + continue + } + h, ok := want[r.Table] + if ok && uint32(r.Mark) == markValue(h.Mark) && uint32(r.Mask) == markMask { + have[r.Table] = true + continue + } + r := r + if err := netlink.RuleDel(&r); err != nil { + return fmt.Errorf("removing stale rule to table %d: %v", r.Table, err) + } + } + for table := TableBase + 1; table <= TableBase+MaxMarks; table++ { + h, ok := want[table] + if !ok { + routes, err := netlink.RouteListFiltered(unix.AF_INET, &netlink.Route{Table: table}, netlink.RT_FILTER_TABLE) + if err != nil { + return err + } + for _, rt := range routes { + rt := rt + if err := netlink.RouteDel(&rt); err != nil { + return fmt.Errorf("removing stale route in table %d: %v", table, err) + } + } + continue + } + if err := netlink.RouteReplace(&netlink.Route{ + LinkIndex: link.Attrs().Index, + Dst: &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)}, + Gw: h.Node.WeaveIP, + Table: table, + Flags: int(netlink.FLAG_ONLINK), + }); err != nil { + return fmt.Errorf("route to holder %s: %v", h.Node.Name, err) + } + if !have[table] { + rule := netlink.NewRule() + rule.Priority = RulePriority + rule.Table = table + rule.Mark = int(markValue(h.Mark)) + rule.Mask = int(markMask) + if err := netlink.RuleAdd(rule); err != nil { + return fmt.Errorf("rule for holder %s: %v", h.Node.Name, err) + } + } + } + return nil +} diff --git a/lb/model.go b/lb/model.go new file mode 100644 index 00000000..df208711 --- /dev/null +++ b/lb/model.go @@ -0,0 +1,308 @@ +// 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 +} diff --git a/lb/model_test.go b/lb/model_test.go new file mode 100644 index 00000000..4cc965d1 --- /dev/null +++ b/lb/model_test.go @@ -0,0 +1,241 @@ +package lb + +import ( + "fmt" + "net" + "strings" + "testing" + + coreapi "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func mac(s string) net.HardwareAddr { + m, err := net.ParseMAC(s) + if err != nil { + panic(err) + } + return m +} + +var ( + cp = Node{Name: "cp", Addresses: []net.IP{net.ParseIP("86.107.168.24").To4()}, + WeaveIP: net.ParseIP("10.32.0.1").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:01")} + w1 = Node{Name: "w1", Addresses: []net.IP{net.ParseIP("86.107.168.15").To4()}, + WeaveIP: net.ParseIP("10.40.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:02")} + w2 = Node{Name: "w2", Addresses: []net.IP{net.ParseIP("86.107.168.88").To4()}, + WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:03")} + nodes = []Node{w2, cp, w1} +) + +func svc(name, ip string, mutate ...func(*coreapi.Service)) *coreapi.Service { + class := Class + s := &coreapi.Service{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name}, + Spec: coreapi.ServiceSpec{ + Type: coreapi.ServiceTypeLoadBalancer, + LoadBalancerClass: &class, + LoadBalancerIP: ip, + Ports: []coreapi.ServicePort{ + {Name: "ssh", Protocol: coreapi.ProtocolTCP, Port: 22}, + {Name: "https", Port: 443}, // protocol defaults to TCP + }, + }, + } + for _, m := range mutate { + m(s) + } + return s +} + +func shows(ip string) func(*coreapi.Service) { + return func(s *coreapi.Service) { + s.Status.LoadBalancer.Ingress = []coreapi.LoadBalancerIngress{{IP: ip}} + } +} + +func ptr[T any](v T) *T { return &v } + +func slice(svcName string, eps map[string]bool, ports map[string]int32) *discovery.EndpointSlice { + sl := &discovery.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: svcName + "-x", Labels: map[string]string{serviceNameLabel: svcName}}, + AddressType: discovery.AddressTypeIPv4, + } + for ip, ready := range eps { + sl.Endpoints = append(sl.Endpoints, discovery.Endpoint{Addresses: []string{ip}, Conditions: discovery.EndpointConditions{Ready: ptr(ready)}}) + } + for name, port := range ports { + sl.Ports = append(sl.Ports, discovery.EndpointPort{Name: ptr(name), Protocol: ptr(coreapi.ProtocolTCP), Port: ptr(port)}) + } + return sl +} + +func TestHolderForwardsToReadyEndpointsAndWritesStatus(t *testing.T) { + s := svc("gitea", "86.107.168.15") + sl := slice("gitea", map[string]bool{"10.40.0.7": true, "10.44.0.3": true, "10.44.0.9": false}, + map[string]int32{"ssh": 2222, "https": 3000}) + d := Compute("w1", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl}) + if len(d.Problems) != 0 { + t.Fatalf("problems: %v", d.Problems) + } + if len(d.Forwards) != 1 || !d.Forwards[0].Address.Equal(net.ParseIP("86.107.168.15")) { + t.Fatalf("forwards: %+v", d.Forwards) + } + f := d.Forwards[0] + if len(f.Ports) != 2 || f.Ports[0].Port != 22 || f.Ports[1].Port != 443 || f.Ports[1].Protocol != coreapi.ProtocolTCP { + t.Fatalf("ports: %+v", f.Ports) + } + if got := f.Ports[0].Endpoints; len(got) != 2 || got[0].String() != "10.40.0.7:2222" || got[1].String() != "10.44.0.3:2222" { + t.Fatalf("ssh endpoints (not-ready one must be left out): %v", got) + } + if got := f.Ports[1].Endpoints; len(got) != 2 || got[0].Port != 3000 { + t.Fatalf("https endpoints: %v", got) + } + if len(d.Status) != 1 || !d.Status[0].Address.Equal(net.ParseIP("86.107.168.15")) { + t.Fatalf("status: %+v", d.Status) + } + if len(d.Holders) != 0 { + t.Fatalf("a holder does not steer its own replies: %+v", d.Holders) + } + // already shown: nothing to write + d = Compute("w1", nodes, []*coreapi.Service{svc("gitea", "86.107.168.15", shows("86.107.168.15"))}, nil) + if len(d.Status) != 0 { + t.Fatalf("status rewritten: %+v", d.Status) + } +} + +func TestOtherNodesSteerRepliesToTheHolderAndNeverWrite(t *testing.T) { + s := svc("gitea", "86.107.168.15") + for _, self := range []string{"cp", "w2"} { + d := Compute(self, nodes, []*coreapi.Service{s}, nil) + if len(d.Forwards) != 0 || len(d.Status) != 0 { + t.Fatalf("%s: only the holder forwards and writes: %+v", self, d) + } + if len(d.Holders) != 1 || d.Holders[0].Node.Name != "w1" || d.Holders[0].Mark < 1 || d.Holders[0].Mark > MaxMarks { + t.Fatalf("%s: holders: %+v", self, d.Holders) + } + } +} + +func TestUnheldOrContestedAddressFailsClosed(t *testing.T) { + // nobody holds it: no forward anywhere; the stale status is cleared by exactly one node (first by name) + s := svc("gitea", "86.107.168.99", shows("86.107.168.99")) + for _, self := range []string{"cp", "w1", "w2"} { + d := Compute(self, nodes, []*coreapi.Service{s}, nil) + if len(d.Forwards) != 0 || len(d.Holders) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "no node holds") { + t.Fatalf("%s: %+v", self, d) + } + clears := len(d.Status) == 1 && d.Status[0].Address == nil + if clears != (self == "cp") { + t.Fatalf("%s: clears=%v", self, clears) + } + } + // two nodes claim it: neither serves it, its holder's view clears the status + both := w2 + both.Addresses = append([]net.IP{net.ParseIP("86.107.168.15").To4()}, w2.Addresses...) + d := Compute("w1", []Node{cp, w1, both}, []*coreapi.Service{svc("gitea", "86.107.168.15", shows("86.107.168.15"))}, nil) + if len(d.Forwards) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "several nodes") { + t.Fatalf("contested: %+v", d) + } + // a bad request is reported, not guessed at + d = Compute("w1", nodes, []*coreapi.Service{svc("gitea", "")}, nil) + if len(d.Forwards) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "spec.loadBalancerIP") { + t.Fatalf("no address: %+v", d) + } +} + +func TestOnlyOptedInLoadBalancersAreServed(t *testing.T) { + other := svc("a", "86.107.168.15", func(s *coreapi.Service) { s.Spec.LoadBalancerClass = ptr("example.com/other") }) + noClass := svc("b", "86.107.168.15", func(s *coreapi.Service) { s.Spec.LoadBalancerClass = nil }) + clusterIP := svc("c", "86.107.168.15", func(s *coreapi.Service) { + s.Spec.Type = coreapi.ServiceTypeClusterIP + }) + d := Compute("w1", nodes, []*coreapi.Service{other, noClass, clusterIP}, nil) + if len(d.Forwards)+len(d.Status)+len(d.Holders)+len(d.Problems) != 0 { + t.Fatalf("touched a Service that did not opt in: %+v", d) + } +} + +func TestParseNodeAndAnnotationsRoundTrip(t *testing.T) { + l := Local{Addresses: []net.IP{net.ParseIP("86.107.168.88").To4(), net.ParseIP("86.107.168.15").To4()}, + WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:03")} + a := l.Annotations() + if a[AnnAddresses] != "86.107.168.15,86.107.168.88" || a[AnnWeave] != "10.44.0.0,aa:aa:aa:aa:aa:03" { + t.Fatalf("annotations: %v", a) + } + n, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "w2", Annotations: a}}) + if err != nil || !n.Holds(net.ParseIP("86.107.168.15")) || !n.WeaveIP.Equal(l.WeaveIP) || n.WeaveMAC.String() != "aa:aa:aa:aa:aa:03" { + t.Fatalf("parsed %+v, %v", n, err) + } + for _, bad := range []map[string]string{{AnnAddresses: "1.2.3.4,nope"}, {AnnWeave: "10.44.0.0"}, {AnnWeave: "x,aa:aa:aa:aa:aa:03"}} { + if _, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "w2", Annotations: bad}}); err == nil { + t.Fatalf("accepted %v", bad) + } + } + if n, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "new"}}); err != nil || len(n.Addresses) != 0 || n.WeaveIP != nil { + t.Fatalf("a node that published nothing holds nothing: %+v %v", n, err) + } +} + +func TestMarksAreDistinctAndStable(t *testing.T) { + var ns []Node + for _, name := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} { + ns = append(ns, Node{Name: name}) + } + first := assignMarks(ns) + seen := map[uint32]bool{} + for _, h := range first { + if seen[h.Mark] || h.Mark < 1 || h.Mark > MaxMarks { + t.Fatalf("bad or repeated mark: %+v", first) + } + seen[h.Mark] = true + } + again := assignMarks(ns) + for i := range first { + if first[i].Mark != again[i].Mark { + t.Fatalf("marks moved between passes") + } + } +} + +func TestRenderedRules(t *testing.T) { + _, podNet, _ := net.ParseCIDR("10.32.0.0/12") + s := svc("gitea", "86.107.168.15") + sl := slice("gitea", map[string]bool{"10.40.0.7": true, "10.44.0.3": true, "10.44.0.4": true}, map[string]int32{"ssh": 2222, "https": 3000}) + d := Compute("w1", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl}) + r := Render(d, podNet, []string{"REWEAVE-LBS-DEADBEEF0000"}) + for _, want := range []string{ + "*nat\n:REWEAVE-LB - [0:0]\n", + ":REWEAVE-LBS-DEADBEEF0000 - [0:0]\n", // stale chain emptied + "-A REWEAVE-LB -d 86.107.168.15/32 -p tcp -m tcp --dport 22 -m comment --comment \"gitea/gitea\" -j REWEAVE-LBS-", + "-m statistic --mode random --probability 0.3333333333 -j DNAT --to-destination 10.40.0.7:2222\n", + "-m statistic --mode random --probability 0.5000000000 -j DNAT --to-destination 10.44.0.3:2222\n", + "-j DNAT --to-destination 10.44.0.4:2222\n", + "COMMIT\n", + } { + if !strings.Contains(r.Nat, want) { + t.Fatalf("nat lacks %q:\n%s", want, r.Nat) + } + } + if strings.Contains(r.Nat, "SNAT") || strings.Contains(r.Nat, "MASQUERADE") { + t.Fatalf("the client's address must survive:\n%s", r.Nat) + } + if r.Mangle != "*mangle\n:REWEAVE-LB-MARK - [0:0]\nCOMMIT\n" { + t.Fatalf("a holder marks nothing:\n%s", r.Mangle) + } + d = Compute("w2", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl}) + r = Render(d, podNet, nil) + m := d.Holders[0].Mark << 24 + for _, want := range []string{ + "-A REWEAVE-LB-MARK -m conntrack --ctdir REPLY -m connmark ! --mark 0x0/0xff000000 -j CONNMARK --restore-mark --nfmask 0xff000000 --ctmask 0xff000000\n", + "-A REWEAVE-LB-MARK -i weave -m mac --mac-source aa:aa:aa:aa:aa:02 ! -s 10.32.0.0/12 -d 10.32.0.0/12 -m conntrack --ctstate NEW -m comment --comment \"holder w1\" -j CONNMARK --set-xmark " + fmt.Sprintf("0x%x", m) + "/0xff000000\n", + } { + if !strings.Contains(r.Mangle, want) { + t.Fatalf("mangle lacks %q:\n%s", want, r.Mangle) + } + } + if r.Nat != "*nat\n:REWEAVE-LB - [0:0]\nCOMMIT\n" { + t.Fatalf("a non-holder forwards nothing:\n%s", r.Nat) + } +} diff --git a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml index fc27c385..6c8c873c 100644 --- a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml +++ b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml @@ -48,6 +48,36 @@ items: verbs: - patch - update + # weave-npc's LoadBalancer agent (lb): Services of class reweave.levk.dev/node-address + - apiGroups: + - '' + resources: + - services + verbs: + - get + - list + - watch + - apiGroups: + - '' + resources: + - services/status + verbs: + - patch + - apiGroups: + - discovery.k8s.io + resources: + - endpointslices + verbs: + - get + - list + - watch + # each node publishes the addresses it holds and its weave bridge (annotations) + - apiGroups: + - '' + resources: + - nodes + verbs: + - patch - apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index 78c0d104..babb0ff0 100644 --- a/prog/weave-npc/main.go +++ b/prog/weave-npc/main.go @@ -21,6 +21,7 @@ import ( "github.com/rajch/weave/common" "github.com/rajch/weave/common/chains" + "github.com/rajch/weave/lb" "github.com/rajch/weave/net" "github.com/rajch/weave/net/ipset" "github.com/rajch/weave/npc" @@ -36,6 +37,7 @@ var ( nodeName string maxList int bridgePortName string + serveLB bool ) func handleError(err error) { @@ -337,6 +339,11 @@ func root(cmd *cobra.Command, args []string) { go podController.Run(wait.NeverStop) go npController.Run(wait.NeverStop) + if serveLB { + // LoadBalancer Services of class lb.Class: addresses this node holds, client address kept + go lb.NewController(nodeName, client, lb.NewDatapath(ipt)).Run(context.Background()) + } + signals := make(chan os.Signal, 1) signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) common.Log.Fatalf("Exiting: %v", <-signals) @@ -354,6 +361,7 @@ func main() { rootCmd.PersistentFlags().StringVar(&nodeName, "node-name", "", "only generate rules that apply to this node") 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") handleFatal(rootCmd.Execute()) }