lb: keep the client's address past weave's masquerade

Weave's expose rule (nat WEAVE: ! -s <alloc> -d <alloc> -j MASQUERADE)
masquerades everything entering the pod network from outside unless it is
for a local pod (weaver-no-masq-local). Live on the SG cluster a pod on
another node than the holder saw the holder's bridge address instead of
the client's. A new chain, REWEAVE-LB-POST, hooked first in nat
POSTROUTING, lets a connection we DNATed to a held address leave the nat
table with its source untouched; one that comes from a pod (hairpin) is
masqueraded to the holder instead, so its reply returns through the
holder's conntrack.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ

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
+}

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
+}

lb/datapath.go

@@ -0,0 +1,276 @@
+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"
+	// PostChain (nat, from POSTROUTING, ahead of weave's WEAVE) keeps the client's address on what ForwardChain DNATed.
+	PostChain = "REWEAVE-LB-POST"
+	// 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)
+	fmt.Fprintf(&nat, ":%s - [0:0]\n", PostChain)
+	var body bytes.Buffer
+	seen := map[string]bool{}
+	for _, f := range d.Forwards {
+		a := f.Address.String()
+		if seen[a] {
+			continue
+		}
+		seen[a] = true
+		// weave masquerades everything entering the pod network from outside unless it is for a local pod
+		// (its expose rule); a connection we DNATed must keep the client's address, so it leaves the nat
+		// table here. One from a pod (hairpin) is masqueraded to this node, so its reply comes back here.
+		fmt.Fprintf(&body, "-A %s -m conntrack --ctstate DNAT --ctorigdst %s/32 -s %s -j MASQUERADE\n", PostChain, a, podNet)
+		fmt.Fprintf(&body, "-A %s -m conntrack --ctstate DNAT --ctorigdst %s/32 -j ACCEPT\n", PostChain, a)
+	}
+	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 {
+				// DNAT to an ip:port needs the protocol in the same rule
+				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 -p %s -m statistic --mode random --probability %.10f -j DNAT --to-destination %s\n",
+						c, proto, 1/float64(n-i), e)
+				} else {
+					fmt.Fprintf(&body, "-A %s -p %s -j DNAT --to-destination %s\n", c, proto, 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 and POSTROUTING jump to our chains first (kube-proxy's and weave's must not see
+// held addresses' connections first).
+func (dp *Datapath) hooks() error {
+	for _, h := range []struct{ table, from, chain string }{
+		{"nat", "PREROUTING", ForwardChain},
+		{"nat", "POSTROUTING", PostChain},
+		{"mangle", "PREROUTING", 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, h.from)
+		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, h.from, rule...); err != nil {
+			return err
+		}
+		if err := dp.ipt.Insert(h.table, h.from, 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
+}

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 "<ip>,<mac>" 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 \"<ip>,<mac>\", 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 \"<ip>,<mac>\", 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
+}

lb/model_test.go

@@ -0,0 +1,250 @@
+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-LB-POST - [0:0]\n",
+		// the client's address survives weave's masquerade; a pod's own connection (hairpin) is masqueraded here
+		"-A REWEAVE-LB-POST -m conntrack --ctstate DNAT --ctorigdst 86.107.168.15/32 -s 10.32.0.0/12 -j MASQUERADE\n",
+		"-A REWEAVE-LB-POST -m conntrack --ctstate DNAT --ctorigdst 86.107.168.15/32 -j ACCEPT\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-",
+		"-p tcp -m statistic --mode random --probability 0.3333333333 -j DNAT --to-destination 10.40.0.7:2222\n",
+		"-p tcp -m statistic --mode random --probability 0.5000000000 -j DNAT --to-destination 10.44.0.3:2222\n",
+		"-p tcp -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)
+		}
+	}
+	for _, line := range strings.Split(r.Nat, "\n") {
+		// iptables refuses a DNAT to ip:port without the protocol in the same rule
+		if strings.Contains(line, "-j DNAT") && !strings.Contains(line, " -p tcp ") {
+			t.Fatalf("DNAT without its protocol: %q", line)
+		}
+	}
+	if strings.Count(r.Nat, "MASQUERADE") != 1 || strings.Contains(r.Nat, "SNAT") {
+		t.Fatalf("only a pod's hairpin connection may be masqueraded:\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]\n:REWEAVE-LB-POST - [0:0]\nCOMMIT\n" {
+		t.Fatalf("a non-holder forwards nothing:\n%s", r.Nat)
+	}
+}

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:

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())
 }

reweave/test/lb/lb-test.yaml

@@ -0,0 +1,75 @@
+# End-to-end test of weave-npc's LoadBalancer agent (lb). Placeholders are
+# filled by run.sh: ADDR (an address HOLDER holds), HOLDER (that node),
+# REMOTE (another node: its pod's replies must be steered back to HOLDER),
+# OTHER (a third node: a pod there connects to ADDR from inside the cluster).
+apiVersion: v1
+kind: List
+items:
+  - apiVersion: v1
+    kind: Namespace
+    metadata:
+      name: lbtest
+  # /clientip answers with the address the connection came from
+  - apiVersion: v1
+    kind: Pod
+    metadata:
+      name: echo-remote
+      namespace: lbtest
+      labels: {app: echo-remote}
+    spec:
+      nodeName: REMOTE
+      containers:
+        - name: echo
+          image: registry.k8s.io/e2e-test-images/agnhost:2.53
+          args: ["netexec", "--http-port=8080"]
+          ports: [{name: http, containerPort: 8080}]
+          readinessProbe: {httpGet: {path: /healthz, port: 8080}}
+  - apiVersion: v1
+    kind: Pod
+    metadata:
+      name: echo-local
+      namespace: lbtest
+      labels: {app: echo-local}
+    spec:
+      nodeName: HOLDER
+      containers:
+        - name: echo
+          image: registry.k8s.io/e2e-test-images/agnhost:2.53
+          args: ["netexec", "--http-port=8080"]
+          ports: [{name: http, containerPort: 8080}]
+          readinessProbe: {httpGet: {path: /healthz, port: 8080}}
+  - apiVersion: v1
+    kind: Pod
+    metadata:
+      name: client
+      namespace: lbtest
+    spec:
+      nodeName: OTHER
+      containers:
+        - name: client
+          image: registry.k8s.io/e2e-test-images/agnhost:2.53
+          args: ["pause"]
+  - apiVersion: v1
+    kind: Service
+    metadata:
+      name: echo-remote
+      namespace: lbtest
+    spec:
+      type: LoadBalancer
+      loadBalancerClass: reweave.levk.dev/node-address
+      loadBalancerIP: ADDR
+      allocateLoadBalancerNodePorts: false
+      selector: {app: echo-remote}
+      ports: [{name: http, port: 18080, targetPort: http}]
+  - apiVersion: v1
+    kind: Service
+    metadata:
+      name: echo-local
+      namespace: lbtest
+    spec:
+      type: LoadBalancer
+      loadBalancerClass: reweave.levk.dev/node-address
+      loadBalancerIP: ADDR
+      allocateLoadBalancerNodePorts: false
+      selector: {app: echo-local}
+      ports: [{name: http, port: 18081, targetPort: http}]

reweave/test/lb/run.sh

@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+# End-to-end test of weave-npc's LoadBalancer agent against a live cluster.
+#
+#   CP="ssh ... root@<control plane>"  HOLDER_SSH="ssh ... root@<holder>" \
+#   ADDR=<address HOLDER holds> HOLDER=<node> REMOTE=<node> OTHER=<node> \
+#   reweave/test/lb/run.sh
+#
+# CP runs kubectl (admin.conf); HOLDER_SSH reads the holder's iptables. The
+# machine running this script is the outside client. Asserts, in order:
+#   1. each Service's status is exactly [{ip: ADDR, ipMode: Proxy}]
+#   2. kube-proxy installed nothing for ADDR (ipMode Proxy leaves it to us)
+#   3. from outside, both ports answer, and the pod sees the client's own
+#      address: from the holder's own pod, and from a pod on another node
+#      (whose replies must be steered back through the holder)
+#   4. a pod on a third node reaches ADDR too
+#   5. fail closed: naming an address nobody holds clears the status and
+#      stops the forward
+#   6. deleting the Services removes the holder's chains
+# Leaves nothing behind (namespace lbtest is deleted, also on failure).
+set -euo pipefail
+
+: "${CP:?}" "${HOLDER_SSH:?}" "${ADDR:?}" "${HOLDER:?}" "${REMOTE:?}" "${OTHER:?}"
+here=$(cd "$(dirname "$0")" && pwd)
+k() { $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl $*"; }
+fail() { echo "FAIL: $*" >&2; exit 1; }
+ok() { echo "ok: $*"; }
+cleanup() { k delete namespace lbtest --ignore-not-found --wait=false >/dev/null 2>&1 || true; }
+trap cleanup EXIT
+
+# eventually <tries> <command...>: retry once a second
+eventually() {
+    local n=$1; shift
+    for _ in $(seq "$n"); do
+        if "$@" >/dev/null 2>&1; then return 0; fi
+        sleep 1
+    done
+    return 1
+}
+
+sed -e "s/\bADDR\b/$ADDR/g; s/\bHOLDER\b/$HOLDER/g; s/\bREMOTE\b/$REMOTE/g; s/\bOTHER\b/$OTHER/g" \
+    "$here/lb-test.yaml" | $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl apply -f -" >/dev/null
+k -n lbtest wait --for=condition=Ready pod --all --timeout=180s >/dev/null || fail "test pods not ready"
+
+status() { k -n lbtest get svc "$1" -o "jsonpath='{.status.loadBalancer.ingress}'"; }
+for s in echo-remote echo-local; do
+    want="[{\"ip\":\"$ADDR\",\"ipMode\":\"Proxy\"}]"
+    eventually 30 test "$(status $s)" = "$want" || fail "$s status: $(status $s), want $want"
+done
+ok "1. status [{ip: $ADDR, ipMode: Proxy}] on both Services"
+
+if $HOLDER_SSH "iptables-save -t nat | grep -E '^-A KUBE-' | grep -F -- '-d $ADDR/32'"; then
+    fail "kube-proxy has rules for $ADDR"
+fi
+ok "2. kube-proxy installed nothing for $ADDR"
+
+# the client's own address, as the holder sees this machine's SSH session
+me=$($HOLDER_SSH 'echo ${SSH_CLIENT%% *}')
+for port in 18080 18081; do
+    got=""
+    for _ in $(seq 30); do
+        got=$(curl -s --max-time 3 "http://$ADDR:$port/clientip" || true)
+        [ -n "$got" ] && break
+        sleep 1
+    done
+    [ "${got%:*}" = "$me" ] || fail "port $port: pod saw '${got}', want this client ($me)"
+done
+ok "3. from outside, both pods see this client's own address ($me), remote pod's replies steered back"
+
+k -n lbtest exec client -- /agnhost connect "$ADDR:18080" --timeout 5s >/dev/null || fail "pod on $OTHER cannot reach $ADDR:18080"
+ok "4. a pod on $OTHER reaches $ADDR:18080"
+
+k -n lbtest patch svc echo-remote --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"192.0.2.1\"}}'" >/dev/null
+eventually 30 test "$(status echo-remote)" = "" || fail "status not cleared for an unheld address: $(status echo-remote)"
+if curl -s --max-time 3 "http://$ADDR:18080/clientip" >/dev/null; then
+    fail "$ADDR:18080 still forwards after its Service moved to an unheld address"
+fi
+ok "5. unheld address: status cleared, forward gone (fail closed)"
+
+k -n lbtest delete svc echo-remote echo-local >/dev/null
+eventually 30 test "$($HOLDER_SSH "iptables-save -t nat | grep -c '^:REWEAVE-LBS-'" || true)" = 0 \
+    || fail "holder kept per-port chains after the Services went away"
+ok "6. Services deleted: holder's chains removed"
+echo PASS