weave/lb/egress.go

package lb

import (
	"fmt"
	"net"
	"sort"
	"strings"

	coreapi "k8s.io/api/core/v1"
)

// Port blocks: every node gets its own source ports on an EgressService's address, so the holder can route a
// reply back to the member's node by port alone (no state on the holder). The range lies above the default
// ephemeral range (32768-60999), so it never meets the holder's own connections from the address.
const (
	BlockBase  = 61000
	BlockSize  = 512
	BlockCount = 8 // 61000-65095
)

// Admission is a pod the CNI plugin announced at ADD, before the API shows its IP.
type Admission struct {
	Namespace, Name string
}

// Member is one pod on this node that an EgressService selects, and where its outside traffic goes.
type Member struct {
	IP      net.IP
	Pod     string // namespace/name
	Service string // namespace/name of its EgressService (empty if it is in several)
	// Unreachable: its outside traffic is dropped (no holder, no block, several EgressServices, ...)
	Unreachable bool
	Address     net.IP // it leaves from Address:[From, To]
	From, To    int32
	Holder      Node // the node that emits it; Holder.Name == self: emitted here
}

// BlockRoute is, on the holder, the replies to Address:[From, To] going back to Peer.
type BlockRoute struct {
	Address  net.IP
	From, To int32
	Peer     Node
}

// EgressStatusWrite is a status this node writes (the holder, or the first node by name to clear it).
type EgressStatusWrite struct {
	Namespace, Name string
	Status          EgressServiceStatus
}

// EgressDesired is this node's part of every EgressService.
type EgressDesired struct {
	Members   []Member
	Routes    []BlockRoute
	Status    []EgressStatusWrite
	NodeAddrs []net.IP // the cluster's node addresses: members' traffic to them is not "outside"
	Problems  []string
}

// Tunnelled reports whether this node sends or receives anything over the egress tunnel.
func (e EgressDesired) Tunnelled(self string) bool {
	if len(e.Routes) > 0 {
		return true
	}
	for _, m := range e.Members {
		if !m.Unreachable && m.Holder.Name != self {
			return true
		}
	}
	return false
}

// Peers are the nodes this node exchanges tunnelled packets with (sorted by name, deduplicated).
func (e EgressDesired) Peers(self string) []Node {
	seen := map[string]Node{}
	for _, r := range e.Routes {
		seen[r.Peer.Name] = r.Peer
	}
	for _, m := range e.Members {
		if !m.Unreachable && m.Holder.Name != self {
			seen[m.Holder.Name] = m.Holder
		}
	}
	out := make([]Node, 0, len(seen))
	for _, n := range seen {
		out = append(out, n)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}

// AllocateBlocks keeps every valid block a node already has and gives each node without one the lowest free
// block, in name order. A node beyond BlockCount gets none (its members are dropped, and that is reported).
func AllocateBlocks(current []PortBlock, nodeNames []string) []PortBlock {
	present := map[string]bool{}
	for _, n := range nodeNames {
		present[n] = true
	}
	used := map[int]bool{}
	have := map[string]PortBlock{}
	for _, b := range current {
		k := int(b.From-BlockBase) / BlockSize
		valid := present[b.Node] && b.From >= BlockBase && int(b.From-BlockBase)%BlockSize == 0 &&
			k < BlockCount && b.To == b.From+BlockSize-1
		if !valid || used[k] {
			continue
		}
		if _, dup := have[b.Node]; dup {
			continue
		}
		used[k] = true
		have[b.Node] = b
	}
	names := append([]string(nil), nodeNames...)
	sort.Strings(names)
	for _, n := range names {
		if _, ok := have[n]; ok {
			continue
		}
		for k := 0; k < BlockCount; k++ {
			if !used[k] {
				used[k] = true
				from := int32(BlockBase + k*BlockSize)
				have[n] = PortBlock{Node: n, From: from, To: from + BlockSize - 1}
				break
			}
		}
	}
	out := make([]PortBlock, 0, len(have))
	for _, b := range have {
		out = append(out, b)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Node < out[j].Node })
	return out
}

func sameStatus(a, b EgressServiceStatus) bool {
	if a.Host != b.Host || a.Address != b.Address || len(a.Blocks) != len(b.Blocks) {
		return false
	}
	for i := range a.Blocks {
		if a.Blocks[i] != b.Blocks[i] {
			return false
		}
	}
	return true
}

// holderOf is S1's rule for a Service's address: the one node that holds spec.loadBalancerIP.
func holderOf(s *coreapi.Service, nodes []Node) (*Node, net.IP) {
	want := net.ParseIP(s.Spec.LoadBalancerIP).To4()
	if want == nil {
		return nil, nil
	}
	var hs []Node
	for _, n := range nodes {
		if n.Holds(want) {
			hs = append(hs, n)
		}
	}
	if len(hs) != 1 {
		return nil, want
	}
	return &hs[0], want
}

// selects reports whether a Service's selector (non-empty) matches the labels.
func selects(s *coreapi.Service, labels map[string]string) bool {
	if len(s.Spec.Selector) == 0 {
		return false
	}
	for k, v := range s.Spec.Selector {
		if labels[k] != v {
			return false
		}
	}
	return true
}

// ComputeEgress decides this node's part of every EgressService: the statuses it writes, the members it hosts and
// where their traffic goes, and (as a holder) where replies to each node's port block go.
// pods are this node's pods; admitted maps an IP to the pod the CNI plugin gave it (newer than the API may show).
func ComputeEgress(self string, nodes []Node, services []*coreapi.Service, egresses []*EgressService,
	pods []*coreapi.Pod, admitted map[string]Admission) EgressDesired {
	var d EgressDesired
	sorted := append([]Node(nil), nodes...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
	byName := map[string]Node{}
	var names []string
	for _, n := range sorted {
		byName[n.Name] = n
		names = append(names, n.Name)
		d.NodeAddrs = append(d.NodeAddrs, n.Addresses...)
	}
	svcs := map[string]*coreapi.Service{}
	for _, s := range services {
		svcs[s.Namespace+"/"+s.Name] = s
	}
	es := append([]*EgressService(nil), egresses...)
	sort.Slice(es, func(i, j int) bool { return es[i].Namespace+"/"+es[i].Name < es[j].Namespace+"/"+es[j].Name })

	// valid[key]: the EgressService is in force: its Service is served from an address one node holds, and the
	// holder has published that (status) - what members and holder both act on
	valid := map[string]bool{}
	for _, e := range es {
		key := e.Namespace + "/" + e.Name
		var want EgressServiceStatus
		var holder *Node
		var addr net.IP
		s := svcs[key]
		switch {
		case e.Spec.Emit != EmitHolder:
			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: spec.emit must be %q (the only mode so far), got %q", key, EmitHolder, e.Spec.Emit))
		case e.Spec.SourceIPBy != "" && e.Spec.SourceIPBy != SourceIPByLoadBalancerIP:
			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: spec.sourceIPBy must be %q, got %q", key, SourceIPByLoadBalancerIP, e.Spec.SourceIPBy))
		case s == nil:
			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: no Service %s", key, key))
		case !eligible(s):
			d.Problems = append(d.Problems, fmt.Sprintf("egress %s: Service %s is not a LoadBalancer of class %s", key, key, Class))
		default:
			holder, addr = holderOf(s, sorted)
			if holder == nil {
				d.Problems = append(d.Problems, fmt.Sprintf("egress %s: no single node holds %s; members' outside traffic is dropped", key, s.Spec.LoadBalancerIP))
			}
		}
		writer := ""
		if holder != nil {
			want = EgressServiceStatus{Host: holder.Name, Address: addr.String(), Blocks: AllocateBlocks(e.Status.Blocks, names)}
			writer = holder.Name
			if len(want.Blocks) < len(names) {
				d.Problems = append(d.Problems, fmt.Sprintf("egress %s: only %d port blocks for %d nodes; members on nodes without one are dropped", key, BlockCount, len(names)))
			}
		} else if len(names) > 0 {
			writer = names[0]
		}
		if writer == self && !sameStatus(e.Status, want) {
			d.Status = append(d.Status, EgressStatusWrite{Namespace: e.Namespace, Name: e.Name, Status: want})
		}
		// in force only once the status in the API says what holds now (members and holder read the same thing)
		if holder != nil && e.Status.Host == holder.Name && e.Status.Address == addr.String() {
			valid[key] = true
			if holder.Name == self {
				for _, b := range e.Status.Blocks {
					if b.Node == self {
						continue
					}
					peer, ok := byName[b.Node]
					if !ok || peer.WeaveIP == nil {
						d.Problems = append(d.Problems, fmt.Sprintf("egress %s: node %s has no weave bridge published; its members' replies cannot reach it", key, b.Node))
						continue
					}
					d.Routes = append(d.Routes, BlockRoute{Address: addr, From: b.From, To: b.To, Peer: peer})
				}
			}
		}
	}

	// this node's pods, one per IP: the CNI plugin's latest word wins, then live pods over terminating ones
	type candidate struct {
		pod *coreapi.Pod
		ip  string
	}
	byIP := map[string]*coreapi.Pod{}
	podByKey := map[string]*coreapi.Pod{}
	for _, p := range pods {
		podByKey[p.Namespace+"/"+p.Name] = p
	}
	var cands []candidate
	for _, p := range pods {
		if p.Spec.NodeName != self || p.Spec.HostNetwork || p.Status.Phase == coreapi.PodSucceeded || p.Status.Phase == coreapi.PodFailed {
			continue
		}
		if ip := net.ParseIP(p.Status.PodIP).To4(); ip != nil {
			cands = append(cands, candidate{p, ip.String()})
		}
	}
	for ip, a := range admitted {
		if p := podByKey[a.Namespace+"/"+a.Name]; p != nil && p.Spec.NodeName == self && !p.Spec.HostNetwork {
			cands = append(cands, candidate{p, ip})
		}
	}
	for _, c := range cands {
		if a, ok := admitted[c.ip]; ok && (a.Namespace != c.pod.Namespace || a.Name != c.pod.Name) {
			continue // the IP was given to another pod since
		}
		cur := byIP[c.ip]
		if cur == nil || (cur.DeletionTimestamp != nil && c.pod.DeletionTimestamp == nil) ||
			(cur.DeletionTimestamp == nil) == (c.pod.DeletionTimestamp == nil) && c.pod.CreationTimestamp.After(cur.CreationTimestamp.Time) {
			byIP[c.ip] = c.pod
		}
	}
	ips := make([]string, 0, len(byIP))
	for ip := range byIP {
		ips = append(ips, ip)
	}
	sort.Strings(ips)
	for _, ip := range ips {
		p := byIP[ip]
		var in []*EgressService
		for _, e := range es {
			if e.Namespace != p.Namespace {
				continue
			}
			if s := svcs[e.Namespace+"/"+e.Name]; s != nil && selects(s, p.Labels) {
				in = append(in, e)
			}
		}
		if len(in) == 0 {
			continue
		}
		m := Member{IP: net.ParseIP(ip).To4(), Pod: p.Namespace + "/" + p.Name, Unreachable: true}
		if len(in) > 1 {
			keys := make([]string, len(in))
			for i, e := range in {
				keys[i] = e.Namespace + "/" + e.Name
			}
			d.Problems = append(d.Problems, fmt.Sprintf("pod %s is selected by several EgressServices (%s); its outside traffic is dropped", m.Pod, strings.Join(keys, ", ")))
			d.Members = append(d.Members, m)
			continue
		}
		e := in[0]
		m.Service = e.Namespace + "/" + e.Name
		if valid[m.Service] {
			holder := byName[e.Status.Host]
			for _, b := range e.Status.Blocks {
				if b.Node == self && (holder.Name == self || holder.WeaveIP != nil) {
					m.Unreachable = false
					m.Address = net.ParseIP(e.Status.Address).To4()
					m.From, m.To = b.From, b.To
					m.Holder = holder
				}
			}
			if m.Unreachable {
				d.Problems = append(d.Problems, fmt.Sprintf("pod %s: egress %s has no port block for %s (or its holder's bridge is unknown); its outside traffic is dropped", m.Pod, m.Service, self))
			}
		}
		d.Members = append(d.Members, m)
	}
	return d
}