weave/lb/datapath.go

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
}