diff --git a/lb/admit/admit.go b/lb/admit/admit.go new file mode 100644 index 00000000..9cb24361 --- /dev/null +++ b/lb/admit/admit.go @@ -0,0 +1,88 @@ +// Package admit is the protocol between the weave CNI plugin and weave-npc's egress agent: at CNI ADD the +// plugin tells the agent which pod got which IP, and the agent answers once the pod's egress rules are in +// place, so an EgressService member's first packet already leaves from the Service's address (or not at all). +// +// It depends on the standard library only: the plugin binary must not grow a Kubernetes client. +package admit + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" +) + +// SocketPath is where the agent listens (a hostPath directory shared with the host, where the plugin runs). +const SocketPath = "/run/weave-npc/egress.sock" + +// Request is one pod's admission: the IP the plugin just allocated to it. +type Request struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + IP string `json:"ip"` +} + +// Response says whether the pod is an EgressService member (its rules are installed either way when 200). +type Response struct { + Member bool `json:"member"` +} + +// Timeout bounds one admission (the agent may have to fetch the pod from the API server). +const Timeout = 20 * time.Second + +// PodFromArgs reads the pod's namespace and name from CNI_ARGS ("K8S_POD_NAMESPACE=..;K8S_POD_NAME=.."). +func PodFromArgs(args string) (namespace, name string) { + for _, kv := range strings.Split(args, ";") { + k, v, ok := strings.Cut(kv, "=") + if !ok { + continue + } + switch k { + case "K8S_POD_NAMESPACE": + namespace = v + case "K8S_POD_NAME": + name = v + } + } + return namespace, name +} + +// Admit asks the agent at socket to admit req, and returns only once the agent has installed the pod's rules. +func Admit(ctx context.Context, socket string, req Request) (Response, error) { + var resp Response + body, err := json.Marshal(req) + if err != nil { + return resp, err + } + client := &http.Client{Transport: &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "unix", socket) + }, + }} + ctx, cancel := context.WithTimeout(ctx, Timeout) + defer cancel() + hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://weave-npc/admit", bytes.NewReader(body)) + if err != nil { + return resp, err + } + hreq.Header.Set("Content-Type", "application/json") + hresp, err := client.Do(hreq) + if err != nil { + return resp, fmt.Errorf("egress admission (weave-npc at %s): %v", socket, err) + } + defer hresp.Body.Close() + out, _ := io.ReadAll(io.LimitReader(hresp.Body, 1<<16)) + if hresp.StatusCode != http.StatusOK { + return resp, fmt.Errorf("egress admission refused (%s): %s", hresp.Status, strings.TrimSpace(string(out))) + } + if err := json.Unmarshal(out, &resp); err != nil { + return resp, fmt.Errorf("egress admission: bad answer: %v", err) + } + return resp, nil +} diff --git a/lb/controller.go b/lb/controller.go index 9589b8ba..31a00a19 100644 --- a/lb/controller.go +++ b/lb/controller.go @@ -4,42 +4,66 @@ import ( "context" "encoding/json" "fmt" + "net" + "net/http" + "os" + "path/filepath" + "sync" "time" coreapi "k8s.io/api/core/v1" discovery "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" 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/rest" "k8s.io/client-go/tools/cache" "github.com/rajch/weave/common" + "github.com/rajch/weave/lb/admit" ) -// Controller watches Services, EndpointSlices and Nodes and keeps this node's part of the picture true. +// Controller watches Services, EndpointSlices, Nodes, EgressServices and this node's Pods, and keeps this +// node's part of the picture true. It also answers the CNI plugin's egress admissions. type Controller struct { self string client kubernetes.Interface + egress rest.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, slices, nodes, pods, egresses cache.Store + synced []cache.InformerSynced + egressSynced cache.InformerSynced // nil while the EgressService API is absent + kick chan struct{} + ready chan struct{} // closed once the first pass ran + + mu sync.Mutex // one pass at a time (the loop, or an admission) + lastProblems map[string]bool // Services whose status this node wrote since it started (so a restart rewrites ipMode once) written map[types.UID]string + // admitted: IP -> pod the CNI plugin gave it; fetched: pods admitted before the pod informer had them + admitted map[string]Admission + fetched map[string]*coreapi.Pod + at map[string]time.Time + // lastEgress is what was last applied, reused while the EgressService caches are not synced (never wipe + // members' rules on a partial view) + lastEgress EgressDesired } // 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{}} +// NewController runs nothing yet; Run does. egress is the EgressService API client. +func NewController(self string, client kubernetes.Interface, egress rest.Interface, dp *Datapath) *Controller { + return &Controller{self: self, client: client, egress: egress, dp: dp, readLocal: ReadLocal, + kick: make(chan struct{}, 1), ready: make(chan struct{}), lastProblems: map[string]bool{}, + written: map[types.UID]string{}, admitted: map[string]Admission{}, fetched: map[string]*coreapi.Pod{}, + at: map[string]time.Time{}} } func (c *Controller) poke() { @@ -49,26 +73,84 @@ func (c *Controller) poke() { } } +func newInformer(c *Controller, lw cache.ListerWatcher, obj runtime.Object) (cache.Store, cache.Controller) { + return cache.NewInformer(lw, obj, 0, cache.ResourceEventHandlerFuncs{ + AddFunc: func(interface{}) { c.poke() }, + UpdateFunc: func(interface{}, interface{}) { c.poke() }, + DeleteFunc: func(interface{}) { c.poke() }, + }) +} + +// egressAPI reports whether the EgressService API is served (the CRD is installed). +func (c *Controller) egressAPI() (bool, error) { + _, err := c.client.Discovery().ServerResourcesForGroupVersion(GroupVersion.String()) + if err == nil { + return true, nil + } + if apierrors.IsNotFound(err) { + return false, nil + } + return false, err +} + +func (c *Controller) startEgress(ctx context.Context) { + var ctl cache.Controller + c.egresses, ctl = newInformer(c, cache.NewListWatchFromClient(c.egress, EgressServices, "", fields.Everything()), &EgressService{}) + go ctl.Run(ctx.Done()) + c.egressSynced = ctl.HasSynced +} + // 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} { + var sc, ec, nc, pc cache.Controller + c.services, sc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "services", "", fields.Everything()), &coreapi.Service{}) + c.slices, ec = newInformer(c, cache.NewListWatchFromClient(c.client.DiscoveryV1().RESTClient(), "endpointslices", "", fields.Everything()), &discovery.EndpointSlice{}) + c.nodes, nc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "nodes", "", fields.Everything()), &coreapi.Node{}) + c.pods, pc = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "pods", "", + fields.OneTermEqualSelector("spec.nodeName", c.self)), &coreapi.Pod{}) + for _, ctl := range []cache.Controller{sc, ec, nc, pc} { go ctl.Run(ctx.Done()) + c.synced = append(c.synced, ctl.HasSynced) } - if !cache.WaitForCacheSync(ctx.Done(), sc.HasSynced, ec.HasSynced, nc.HasSynced) { + // with the API present, nothing is applied before EgressServices are known: an empty view would wipe + // members' rules (their traffic would leave from the node); without it, there can be no members + for { + present, err := c.egressAPI() + if err == nil { + if present { + c.startEgress(ctx) + c.synced = append(c.synced, c.egressSynced) + } + break + } + common.Log.Warnf("lb: cannot tell whether the EgressService API is served (retrying): %v", err) + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } + go c.serveAdmission(ctx) + if !cache.WaitForCacheSync(ctx.Done(), c.synced...) { return } tick := time.NewTicker(Resync) defer tick.Stop() c.poke() + first := true for { select { case <-ctx.Done(): return case <-tick.C: + if c.egressSynced == nil { + // the API may have been installed since + if present, err := c.egressAPI(); err == nil && present { + c.mu.Lock() + c.startEgress(ctx) + c.mu.Unlock() + } + } case <-c.kick: // let a burst of events settle into one pass time.Sleep(200 * time.Millisecond) @@ -77,22 +159,25 @@ func (c *Controller) Run(ctx context.Context) { default: } } - if err := c.sync(ctx); err != nil { + c.mu.Lock() + err := c.sync(ctx) + c.mu.Unlock() + if err != nil { common.Log.Errorf("lb: %v", err) } + if first { + first = false + close(c.ready) + } } } -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() }, - }) +// egressViewSynced: the EgressService view can be trusted (absent API: trivially empty; else synced). +func (c *Controller) egressViewSynced() bool { + return c.egressSynced == nil || c.egressSynced() } -// sync is one pass: publish this node, compute, apply, write statuses. +// sync is one pass: publish this node, compute, apply, write statuses. Callers hold c.mu. func (c *Controller) sync(ctx context.Context) error { local, err := c.readLocal() if err != nil { @@ -129,8 +214,17 @@ func (c *Controller) sync(ctx context.Context) error { d.Status = append(d.Status, Status{Namespace: f.Namespace, Name: f.Name, Address: f.Address}) } } + + e := c.lastEgress + e.Status = nil // reusing the datapath, never the writes of an earlier view + if c.egressViewSynced() { + e = c.computeEgress(nodes, svcs) + } else { + problems = append(problems, "EgressServices not synced yet: keeping the egress rules last applied") + } + current := map[string]bool{} - for _, p := range append(problems, d.Problems...) { + for _, p := range append(append(problems, d.Problems...), e.Problems...) { current[p] = true if !c.lastProblems[p] { common.Log.Warnf("lb: %s", p) @@ -138,17 +232,67 @@ func (c *Controller) sync(ctx context.Context) error { } c.lastProblems = current - if err := c.dp.Apply(d, local.PodNet); err != nil { + if err := c.dp.Apply(d, e, c.self, local); err != nil { return fmt.Errorf("applying datapath: %v", err) } + c.lastEgress = e for _, st := range d.Status { if err := c.writeStatus(ctx, st); err != nil { return err } } + for _, st := range e.Status { + if err := PatchEgressStatus(ctx, c.egress, st.Namespace, st.Name, st.Status); err != nil { + return fmt.Errorf("status of egress %s/%s: %v", st.Namespace, st.Name, err) + } + common.Log.Infof("lb: egress %s/%s: host %q, address %q, %d port blocks", st.Namespace, st.Name, st.Status.Host, st.Status.Address, len(st.Status.Blocks)) + } return nil } +// computeEgress gathers this node's pods (informer, plus pods admitted before the informer had them) and computes. +func (c *Controller) computeEgress(nodes []Node, svcs []*coreapi.Service) EgressDesired { + var egresses []*EgressService + if c.egresses != nil { + for _, o := range c.egresses.List() { + egresses = append(egresses, o.(*EgressService)) + } + } + seen := map[string]bool{} + var pods []*coreapi.Pod + for _, o := range c.pods.List() { + p := o.(*coreapi.Pod) + seen[p.Namespace+"/"+p.Name] = true + pods = append(pods, p) + } + // forget admissions the API has caught up with or that are long gone + for ip, a := range c.admitted { + key := a.Namespace + "/" + a.Name + o, ok, _ := c.pods.GetByKey(key) + caughtUp := ok && o.(*coreapi.Pod).Status.PodIP == ip + if caughtUp || (!ok && time.Since(c.at[ip]) > 2*time.Minute) { + delete(c.admitted, ip) + delete(c.at, ip) + } + } + for key, p := range c.fetched { + stillAdmitted := false + for _, a := range c.admitted { + if a.Namespace+"/"+a.Name == key { + stillAdmitted = true + } + } + if !stillAdmitted { + delete(c.fetched, key) + continue + } + if !seen[key] { + pods = append(pods, p) + } + } + return ComputeEgress(c.self, nodes, svcs, egresses, pods, c.admitted) +} + func hasStatusFor(d Desired, f Forward) bool { for _, s := range d.Status { if s.Namespace == f.Namespace && s.Name == f.Name { @@ -166,6 +310,95 @@ func (c *Controller) service(ns, name string) *coreapi.Service { return o.(*coreapi.Service) } +// serveAdmission answers the CNI plugin on admit.SocketPath: record the pod's IP, run a pass, answer once the +// pod's rules are in place. A pod is refused (the plugin fails the ADD, kubelet retries) while this node cannot +// yet tell whether it is a member. +func (c *Controller) serveAdmission(ctx context.Context) { + if err := os.MkdirAll(filepath.Dir(admit.SocketPath), 0o700); err != nil { + common.Log.Errorf("lb: admission socket: %v", err) + return + } + _ = os.Remove(admit.SocketPath) + l, err := net.Listen("unix", admit.SocketPath) + if err != nil { + common.Log.Errorf("lb: admission socket: %v", err) + return + } + _ = os.Chmod(admit.SocketPath, 0o600) + mux := http.NewServeMux() + mux.HandleFunc("/admit", func(w http.ResponseWriter, r *http.Request) { + var req admit.Request + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&req); err != nil { + http.Error(w, "bad request: "+err.Error(), http.StatusBadRequest) + return + } + member, code, err := c.admit(r.Context(), req) + if err != nil { + common.Log.Warnf("lb: admission of %s/%s (%s): %v", req.Namespace, req.Name, req.IP, err) + http.Error(w, err.Error(), code) + return + } + _ = json.NewEncoder(w).Encode(admit.Response{Member: member}) + }) + srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { + <-ctx.Done() + _ = srv.Close() + }() + if err := srv.Serve(l); err != nil && err != http.ErrServerClosed { + common.Log.Errorf("lb: admission server: %v", err) + } +} + +func (c *Controller) admit(ctx context.Context, req admit.Request) (bool, int, error) { + ip := net.ParseIP(req.IP).To4() + if req.Namespace == "" || req.Name == "" || ip == nil { + return false, http.StatusBadRequest, fmt.Errorf("need namespace, name and an IPv4 address") + } + select { + case <-c.ready: + case <-ctx.Done(): + return false, http.StatusServiceUnavailable, ctx.Err() + case <-time.After(admit.Timeout - 5*time.Second): + return false, http.StatusServiceUnavailable, fmt.Errorf("not ready yet") + } + key := req.Namespace + "/" + req.Name + var pod *coreapi.Pod + if o, ok, _ := c.pods.GetByKey(key); ok { + pod = o.(*coreapi.Pod) + } else { + gctx, cancel := context.WithTimeout(ctx, 10*time.Second) + p, err := c.client.CoreV1().Pods(req.Namespace).Get(gctx, req.Name, metav1.GetOptions{}) + cancel() + if err != nil { + return false, http.StatusServiceUnavailable, fmt.Errorf("fetching pod: %v", err) + } + pod = p + } + if pod.Spec.NodeName != c.self { + return false, http.StatusConflict, fmt.Errorf("pod is bound to %q, not this node", pod.Spec.NodeName) + } + c.mu.Lock() + defer c.mu.Unlock() + if !c.egressViewSynced() { + return false, http.StatusServiceUnavailable, fmt.Errorf("EgressServices not synced yet") + } + c.admitted[ip.String()] = Admission{Namespace: req.Namespace, Name: req.Name} + c.at[ip.String()] = time.Now() + if _, ok, _ := c.pods.GetByKey(key); !ok { + c.fetched[key] = pod + } + if err := c.sync(ctx); err != nil { + return false, http.StatusServiceUnavailable, err + } + for _, m := range c.lastEgress.Members { + if m.IP.Equal(ip) { + return true, http.StatusOK, nil + } + } + return false, http.StatusOK, nil +} + // 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() diff --git a/lb/datapath.go b/lb/datapath.go index d76ef2ea..5dd6c751 100644 --- a/lb/datapath.go +++ b/lb/datapath.go @@ -44,17 +44,18 @@ func portChain(f Forward, p Port) string { // Rules is the full content of the chains this package owns, in iptables-restore form. type Rules struct { - Nat, Mangle string + Nat, Mangle, Filter 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. +// Render turns the desired state (load-balancer forwards and egress) into iptables-restore input for the nat, +// mangle and filter tables. podNet is the weave allocation range (pods and bridges). // 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 { +func Render(d Desired, e EgressDesired, self string, 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) + for _, c := range []string{ForwardChain, PostChain, ClassChain} { + fmt.Fprintf(&nat, ":%s - [0:0]\n", c) + } var body bytes.Buffer seen := map[string]bool{} for _, f := range d.Forwards { @@ -93,12 +94,12 @@ func Render(d Desired, podNet *net.IPNet, stale []string) Rules { // 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) + for _, c := range []string{MarkChain, MSSChain} { + fmt.Fprintf(&mangle, ":%s - [0:0]\n", c) + } 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", @@ -110,8 +111,19 @@ func Render(d Desired, podNet *net.IPNet, stale []string) Rules { MarkChain, bridge, h.Node.WeaveMAC, podNet, podNet, "holder "+h.Node.Name, markValue(h.Mark), markMask) } } + + var filter bytes.Buffer + filter.WriteString("*filter\n") + for _, c := range []string{FwdChain, InChain} { + fmt.Fprintf(&filter, ":%s - [0:0]\n", c) + } + renderEgress(e, self, podNet, &body, &mangle, &filter) + + nat.Write(body.Bytes()) + nat.WriteString("COMMIT\n") mangle.WriteString("COMMIT\n") - return Rules{Nat: nat.String(), Mangle: mangle.String()} + filter.WriteString("COMMIT\n") + return Rules{Nat: nat.String(), Mangle: mangle.String(), Filter: filter.String()} } // Datapath applies Desired to this node. @@ -133,11 +145,17 @@ func NewDatapath(ipt *iptables.IPTables) *Datapath { }} } -// 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 { +// Apply makes the node's iptables chains, tunnel, policy rules and routes exactly what d and e say. +func (dp *Datapath) Apply(d Desired, e EgressDesired, self string, local Local) error { + podNet := local.PodNet if err := dp.hooks(); err != nil { return err } + // the tunnel device must exist before rules naming it are restored (-o/-i on a missing device is fine for + // iptables, but the routes and tables need it) + if err := ensureTunnel(e.Tunnelled(self), local); err != nil { + return err + } existing, err := dp.ipt.ListChains("nat") if err != nil { return err @@ -154,28 +172,39 @@ func (dp *Datapath) Apply(d Desired, podNet *net.IPNet) error { 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 + r := Render(d, e, self, podNet, stale) + // filter first: the guards are in place before the marks and SNATs that they guard + for _, payload := range []string{r.Filter, r.Nat, r.Mangle} { + if err := dp.restore(payload); 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) + if err := dp.routes(d.Holders); err != nil { + return err + } + return egressNet(e, self, local) } -// hooks makes sure PREROUTING and POSTROUTING jump to our chains first (kube-proxy's and weave's must not see -// held addresses' connections first). +// hooks makes sure the built-in chains jump to ours: first (kube-proxy's and weave's rules must not see these +// connections first), except ClassChain, which must come after kube-proxy's KUBE-SERVICES (whose DNAT ends the +// nat walk, so a connection to a Service is never classed as going outside). func (dp *Datapath) hooks() error { - for _, h := range []struct{ table, from, chain string }{ - {"nat", "PREROUTING", ForwardChain}, - {"nat", "POSTROUTING", PostChain}, - {"mangle", "PREROUTING", MarkChain}, + for _, h := range []struct { + table, from, chain string + last bool + }{ + {"nat", "PREROUTING", ForwardChain, false}, + {"nat", "POSTROUTING", PostChain, false}, + {"nat", "PREROUTING", ClassChain, true}, + {"mangle", "PREROUTING", MarkChain, false}, + {"mangle", "FORWARD", MSSChain, false}, + {"filter", "FORWARD", FwdChain, false}, + {"filter", "INPUT", InChain, false}, } { ok, err := dp.ipt.ChainExists(h.table, h.chain) if err != nil { @@ -191,6 +220,27 @@ func (dp *Datapath) hooks() error { if err != nil { return err } + if h.last { + ours, kube := -1, -1 + for i, r := range rules { + if strings.HasSuffix(r, "-j "+h.chain) { + ours = i + } + if strings.HasSuffix(r, "-j KUBE-SERVICES") { + kube = i + } + } + if ours > kube { + continue + } + if err := dp.ipt.DeleteIfExists(h.table, h.from, rule...); err != nil { + return err + } + if err := dp.ipt.Append(h.table, h.from, rule...); err != nil { + return err + } + continue + } // rules[0] is the policy line; our jump must be rules[1] if len(rules) > 1 && strings.HasSuffix(rules[1], "-j "+h.chain) { continue diff --git a/lb/egress.go b/lb/egress.go new file mode 100644 index 00000000..ff5c08ed --- /dev/null +++ b/lb/egress.go @@ -0,0 +1,340 @@ +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 +} diff --git a/lb/egress_api.go b/lb/egress_api.go new file mode 100644 index 00000000..2b3526a4 --- /dev/null +++ b/lb/egress_api.go @@ -0,0 +1,144 @@ +package lb + +import ( + "context" + "encoding/json" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/rest" +) + +// EgressService (reweave.levk.dev/v1alpha1) makes the pods a LoadBalancer Service selects leave the cluster +// from the Service's address. It is named after the Service (same namespace, same name). Its semantics follow +// OVN-Kubernetes' EgressService (k8s.ovn.org/v1, sourceIPBy: LoadBalancerIP) where they agree; they differ in +// that membership is every pod the Service's selector matches (by pod IP, never by readiness), the holder is +// the node that holds the address (no election), and a member's traffic to the outside leaves as the address +// or not at all. +type EgressService struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec EgressServiceSpec `json:"spec"` + Status EgressServiceStatus `json:"status,omitempty"` +} + +// EgressServiceSpec is what the operator asks for. +type EgressServiceSpec struct { + // SourceIPBy is LoadBalancerIP: members leave from the Service's load-balancer address (the only mode in v1). + SourceIPBy string `json:"sourceIPBy,omitempty"` + // Emit is where members' packets leave the cluster. Explicit, no default: Holder (tunnelled to the node + // holding the address, which emits them) is the only mode in v1. + Emit string `json:"emit"` +} + +// EgressServiceStatus is written by the holder alone. +type EgressServiceStatus struct { + // Host is the node that holds the address and emits members' packets. + Host string `json:"host,omitempty"` + // Address is the address members leave from. + Address string `json:"address,omitempty"` + // Blocks gives every node its own source ports on Address, so the holder routes replies back by port. + Blocks []PortBlock `json:"blocks,omitempty"` +} + +// PortBlock is the source-port range [From, To] on the address that one node's members use. +type PortBlock struct { + Node string `json:"node"` + From int32 `json:"from"` + To int32 `json:"to"` +} + +// EgressServiceList is a list of EgressServices. +type EgressServiceList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []EgressService `json:"items"` +} + +const ( + // Group and Version of the EgressService API. + Group = "reweave.levk.dev" + Version = "v1alpha1" + // EgressServices is the resource name. + EgressServices = "egressservices" + + SourceIPByLoadBalancerIP = "LoadBalancerIP" + EmitHolder = "Holder" +) + +// GroupVersion of the EgressService API. +var GroupVersion = schema.GroupVersion{Group: Group, Version: Version} + +// DeepCopyObject implements runtime.Object. +func (e *EgressService) DeepCopyObject() runtime.Object { return e.DeepCopy() } + +// DeepCopy copies the whole object. +func (e *EgressService) DeepCopy() *EgressService { + if e == nil { + return nil + } + out := *e + e.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Status.Blocks = append([]PortBlock(nil), e.Status.Blocks...) + return &out +} + +// DeepCopyObject implements runtime.Object. +func (l *EgressServiceList) DeepCopyObject() runtime.Object { + if l == nil { + return nil + } + out := *l + l.ListMeta.DeepCopyInto(&out.ListMeta) + out.Items = make([]EgressService, len(l.Items)) + for i := range l.Items { + out.Items[i] = *l.Items[i].DeepCopy() + } + return &out +} + +func addKnownTypes(s *runtime.Scheme) error { + s.AddKnownTypes(GroupVersion, &EgressService{}, &EgressServiceList{}) + metav1.AddToGroupVersion(s, GroupVersion) + return nil +} + +// NewEgressClient is a REST client for the EgressService API (no generated clientset: two plain types). +func NewEgressClient(base *rest.Config) (*rest.RESTClient, error) { + scheme := runtime.NewScheme() + if err := addKnownTypes(scheme); err != nil { + return nil, err + } + cfg := *base + cfg.GroupVersion = &GroupVersion + cfg.APIPath = "/apis" + cfg.ContentType = runtime.ContentTypeJSON + cfg.NegotiatedSerializer = serializer.NewCodecFactory(scheme).WithoutConversion() + if cfg.UserAgent == "" { + cfg.UserAgent = rest.DefaultKubernetesUserAgent() + } + return rest.RESTClientFor(&cfg) +} + +// PatchEgressStatus merge-patches an EgressService's status (nil fields are removed). +func PatchEgressStatus(ctx context.Context, c rest.Interface, namespace, name string, st EgressServiceStatus) error { + body := map[string]interface{}{"host": nil, "address": nil, "blocks": nil} + if st.Host != "" { + body["host"] = st.Host + } + if st.Address != "" { + body["address"] = st.Address + } + if len(st.Blocks) > 0 { + body["blocks"] = st.Blocks + } + patch, err := json.Marshal(map[string]interface{}{"status": body}) + if err != nil { + return err + } + return c.Patch(types.MergePatchType).Namespace(namespace).Resource(EgressServices).Name(name). + SubResource("status").Body(patch).Do(ctx).Error() +} diff --git a/lb/egress_datapath.go b/lb/egress_datapath.go new file mode 100644 index 00000000..49fd8e88 --- /dev/null +++ b/lb/egress_datapath.go @@ -0,0 +1,361 @@ +package lb + +import ( + "bytes" + "fmt" + "net" + "os" + "sort" + + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +const ( + // ClassChain (nat, appended to PREROUTING after kube-proxy's KUBE-SERVICES) marks members' new connections to + // the outside. kube-proxy's DNAT ends the nat walk, so a connection to a Service never gets here. + ClassChain = "REWEAVE-LB-CLASS" + // FwdChain (filter, first in FORWARD) drops members' outside packets that would not leave as their address. + FwdChain = "REWEAVE-LB-FWD" + // InChain (filter, first in INPUT) accepts the egress tunnel only from the cluster's nodes. + InChain = "REWEAVE-LB-IN" + // MSSChain (mangle, first in FORWARD) clamps TCP MSS into the tunnel. + MSSChain = "REWEAVE-LB-MSS" + + // Tunnel is the egress tunnel: ipip over weave between nodes' weave bridges, no fixed remote (each route names it). + Tunnel = "reweave-eg" + // EgressMark marks a member's packets to the outside (connmark on its connections, fwmark on each packet). + EgressMark = uint32(0x100000) + + // HolderRulePriority: replies to a block go to its node; ahead of the local table, which moves to LocalRulePriority, + // since the address is usually local on the holder. + HolderRulePriority = 90 + LocalRulePriority = 100 + // MemberRulePriority: a member's marked packets go to its holder (or nowhere). + MemberRulePriority = 9100 + // EgressTableBase + peer index routes into the tunnel towards that peer; EgressTableBase itself is unreachable. + EgressTableBase = 53000 + + ipipOverhead = 20 +) + +// renderEgress appends the egress rules of each table to the bodies of S1's payloads. +func renderEgress(e EgressDesired, self string, podNet *net.IPNet, nat, mangle, filter *bytes.Buffer) { + members := e.Members + if len(members) > 0 { + // what is not "outside": the pod network, multicast, the cluster's nodes + fmt.Fprintf(nat, "-A %s -d %s -j RETURN\n", ClassChain, podNet) + fmt.Fprintf(nat, "-A %s -d 224.0.0.0/4 -j RETURN\n", ClassChain) + seen := map[string]bool{} + for _, a := range e.NodeAddrs { + if !seen[a.String()] { + seen[a.String()] = true + fmt.Fprintf(nat, "-A %s -d %s/32 -j RETURN\n", ClassChain, a) + } + } + for _, m := range members { + fmt.Fprintf(nat, "-A %s -s %s/32 -m comment --comment %q -j CONNMARK --set-xmark 0x%x/0x%x\n", ClassChain, m.IP, m.Pod, EgressMark, EgressMark) + fmt.Fprintf(nat, "-A %s -s %s/32 -j MARK --set-xmark 0x%x/0x%x\n", ClassChain, m.IP, EgressMark, EgressMark) + } + // later packets of a member's outside connection carry the mark too (the nat table saw only the first) + fmt.Fprintf(mangle, "-A %s -m conntrack --ctdir ORIGINAL -m connmark --mark 0x%x/0x%x -j MARK --set-xmark 0x%x/0x%x\n", + MarkChain, EgressMark, EgressMark, EgressMark, EgressMark) + } + for _, m := range members { + if m.Unreachable { + fmt.Fprintf(filter, "-A %s -s %s/32 -m mark --mark 0x%x/0x%x -m comment --comment %q -j DROP\n", FwdChain, m.IP, EgressMark, EgressMark, m.Pod) + continue + } + out := "-o " + Tunnel + if m.Holder.Name == self { + // emitted here, from the address: never into the pod network unSNATed + out = "! -o " + bridge + fmt.Fprintf(filter, "-A %s -s %s/32 -o %s -m mark --mark 0x%x/0x%x -j DROP\n", FwdChain, m.IP, bridge, EgressMark, EgressMark) + } + for _, proto := range []string{"tcp", "udp"} { + fmt.Fprintf(nat, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -p %s -j SNAT --to-source %s:%d-%d\n", + PostChain, m.IP, out, EgressMark, EgressMark, proto, m.Address, m.From, m.To) + fmt.Fprintf(filter, "-A %s -s %s/32 %s -m mark --mark 0x%x/0x%x -p %s -j RETURN\n", FwdChain, m.IP, out, EgressMark, EgressMark, proto) + } + // anything else a member sends outside (other protocols, other ways out) does not leave + fmt.Fprintf(filter, "-A %s -s %s/32 -m mark --mark 0x%x/0x%x -m comment --comment %q -j DROP\n", FwdChain, m.IP, EgressMark, EgressMark, m.Pod) + } + if !e.Tunnelled(self) { + return + } + // tunnel arrivals: replies to members here (established), members' packets from their blocks (as holder) + addrs := map[string]bool{} + for _, m := range members { + if !m.Unreachable && m.Holder.Name != self && !addrs[m.Address.String()] { + addrs[m.Address.String()] = true + // still addressed to the address after nat: no member connection matched it + fmt.Fprintf(filter, "-A %s -i %s -d %s/32 -j DROP\n", FwdChain, Tunnel, m.Address) + } + } + fmt.Fprintf(filter, "-A %s -i %s -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN\n", FwdChain, Tunnel) + for _, r := range e.Routes { + for _, proto := range []string{"tcp", "udp"} { + fmt.Fprintf(filter, "-A %s -i %s -s %s/32 -p %s -m %s --sport %d:%d -m comment --comment %q -j RETURN\n", + FwdChain, Tunnel, r.Address, proto, proto, r.From, r.To, "block of "+r.Peer.Name) + } + } + fmt.Fprintf(filter, "-A %s -i %s -j DROP\n", FwdChain, Tunnel) + // the tunnel itself only from the cluster's nodes' bridges + for _, p := range e.Peers(self) { + fmt.Fprintf(filter, "-A %s -p 4 -s %s/32 -m mac --mac-source %s -j RETURN\n", InChain, p.WeaveIP, p.WeaveMAC) + } + fmt.Fprintf(filter, "-A %s -p 4 -j DROP\n", InChain) + fmt.Fprintf(mangle, "-A %s -o %s -p tcp -m tcp --tcp-flags SYN,RST SYN -j TCPMSS --clamp-mss-to-pmtu\n", MSSChain, Tunnel) +} + +// peerTables gives each tunnel peer its routing table (EgressTableBase + a stable index). +func peerTables(e EgressDesired, self string) map[string]int { + out := map[string]int{} + for _, h := range assignMarks(e.Peers(self)) { + out[h.Node.Name] = EgressTableBase + int(h.Mark) + } + return out +} + +// egressRule is a policy rule this package owns, in a comparable form. +type egressRule struct { + priority int + src, dst string + ipproto int + from, to uint16 + mark uint32 + table int +} + +func (r egressRule) key() string { + return fmt.Sprintf("%d|%s|%s|%d|%d-%d|%x|%d", r.priority, r.src, r.dst, r.ipproto, r.from, r.to, r.mark, r.table) +} + +// wantRules is every egress policy rule this node needs. +func wantRules(e EgressDesired, self string) []egressRule { + tables := peerTables(e, self) + var out []egressRule + for _, r := range e.Routes { + for _, proto := range []int{unix.IPPROTO_TCP, unix.IPPROTO_UDP} { + out = append(out, egressRule{priority: HolderRulePriority, dst: r.Address.String() + "/32", ipproto: proto, + from: uint16(r.From), to: uint16(r.To), table: tables[r.Peer.Name]}) + } + } + for _, m := range e.Members { + if !m.Unreachable && m.Holder.Name == self { + continue // emitted here: the main table already sends it out + } + table := EgressTableBase + if !m.Unreachable { + table = tables[m.Holder.Name] + } + out = append(out, egressRule{priority: MemberRulePriority, src: m.IP.String() + "/32", mark: EgressMark, table: table}) + } + sort.Slice(out, func(i, j int) bool { return out[i].key() < out[j].key() }) + return out +} + +func ruleOf(r netlink.Rule) egressRule { + er := egressRule{priority: r.Priority, ipproto: r.IPProto, table: r.Table} + if r.Src != nil { + er.src = r.Src.String() + } + if r.Dst != nil { + er.dst = r.Dst.String() + } + if r.Dport != nil { + er.from, er.to = r.Dport.Start, r.Dport.End + } + if r.Mark > 0 { + er.mark = uint32(r.Mark) + } + if er.ipproto < 0 { + er.ipproto = 0 + } + return er +} + +// egressNet makes the tunnel, the local rule's place, the policy rules and the tunnel tables exactly what e needs. +func egressNet(e EgressDesired, self string, local Local) error { + tunnelled := e.Tunnelled(self) // the device itself is made first, by Apply (ensureTunnel) + if len(e.Routes) > 0 { + if err := moveLocalRule(); err != nil { + return err + } + } + tables := peerTables(e, self) + peers := map[int]Node{} + for _, p := range e.Peers(self) { + peers[tables[p.Name]] = p + } + unreachable := false + for _, m := range e.Members { + unreachable = unreachable || m.Unreachable + } + var link netlink.Link + if tunnelled { + l, err := netlink.LinkByName(Tunnel) + if err != nil { + return fmt.Errorf("egress tunnel: %v", err) + } + link = l + } + for table := EgressTableBase; table <= EgressTableBase+MaxMarks; table++ { + var want *netlink.Route + if p, ok := peers[table]; ok && link != nil { + want = &netlink.Route{LinkIndex: link.Attrs().Index, Dst: &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)}, + Gw: p.WeaveIP, Table: table, Flags: int(netlink.FLAG_ONLINK)} + } else if table == EgressTableBase && unreachable { + want = &netlink.Route{Dst: &net.IPNet{IP: net.IPv4zero, Mask: net.CIDRMask(0, 32)}, Table: table, Type: unix.RTN_UNREACHABLE} + } + routes, err := netlink.RouteListFiltered(unix.AF_INET, &netlink.Route{Table: table}, netlink.RT_FILTER_TABLE) + if err != nil { + return err + } + for _, rt := range routes { + if want != nil && rt.Type == want.Type && rt.LinkIndex == want.LinkIndex && rt.Gw.Equal(want.Gw) { + continue + } + rt := rt + if err := netlink.RouteDel(&rt); err != nil { + return fmt.Errorf("removing stale route in table %d: %v", table, err) + } + } + if want != nil { + if err := netlink.RouteReplace(want); err != nil { + return fmt.Errorf("route in table %d: %v", table, err) + } + } + } + want := map[string]egressRule{} + for _, r := range wantRules(e, self) { + want[r.key()] = r + } + rules, err := netlink.RuleList(unix.AF_INET) + if err != nil { + return err + } + have := map[string]bool{} + for _, r := range rules { + if r.Priority != HolderRulePriority && r.Priority != MemberRulePriority { + continue + } + k := ruleOf(r).key() + if _, ok := want[k]; ok && !have[k] { + have[k] = true + continue + } + r := r + if err := netlink.RuleDel(&r); err != nil { + return fmt.Errorf("removing stale egress rule: %v", err) + } + } + for k, r := range want { + if have[k] { + continue + } + rule := netlink.NewRule() + rule.Priority = r.priority + rule.Table = r.table + if r.src != "" { + _, rule.Src, _ = net.ParseCIDR(r.src) + } + if r.dst != "" { + _, rule.Dst, _ = net.ParseCIDR(r.dst) + } + if r.ipproto != 0 { + rule.IPProto = r.ipproto + rule.Dport = netlink.NewRulePortRange(r.from, r.to) + } + if r.mark != 0 { + rule.Mark = int(r.mark) + rule.Mask = int(r.mark) + } + if err := netlink.RuleAdd(rule); err != nil { + return fmt.Errorf("adding egress rule %s: %v", k, err) + } + } + return nil +} + +// moveLocalRule puts the local table's lookup after HolderRulePriority (it is at 0 by default), so the holder can +// route replies to its own address onwards to their node instead of delivering them to itself. +func moveLocalRule() error { + rules, err := netlink.RuleList(unix.AF_INET) + if err != nil { + return err + } + var at0, atLocal bool + for _, r := range rules { + if r.Table == unix.RT_TABLE_LOCAL && r.Src == nil && r.Dst == nil && r.Mark <= 0 { + switch r.Priority { + case 0: + at0 = true + case LocalRulePriority: + atLocal = true + } + } + } + if !atLocal { + rule := netlink.NewRule() + rule.Priority = LocalRulePriority + rule.Table = unix.RT_TABLE_LOCAL + if err := netlink.RuleAdd(rule); err != nil { + return fmt.Errorf("adding local table rule at %d: %v", LocalRulePriority, err) + } + } + if at0 { + rule := netlink.NewRule() + rule.Priority = 0 + rule.Table = unix.RT_TABLE_LOCAL + if err := netlink.RuleDel(rule); err != nil { + return fmt.Errorf("removing local table rule at 0: %v", err) + } + } + return nil +} + +// ensureTunnel creates (or removes) the egress tunnel: ipip with this node's bridge address as its local end. +func ensureTunnel(want bool, local Local) error { + link, err := netlink.LinkByName(Tunnel) + exists := err == nil + if !want { + if exists { + return netlink.LinkDel(link) + } + return nil + } + if local.WeaveIP == nil { + return fmt.Errorf("egress tunnel needs the weave bridge's address") + } + br, err := netlink.LinkByName(bridge) + if err != nil { + return fmt.Errorf("weave bridge: %v", err) + } + mtu := br.Attrs().MTU - ipipOverhead + if exists { + t, ok := link.(*netlink.Iptun) + if !ok || !t.Local.Equal(local.WeaveIP) || t.Remote != nil && !t.Remote.IsUnspecified() || t.Attrs().MTU != mtu { + if err := netlink.LinkDel(link); err != nil { + return err + } + exists = false + } + } + if !exists { + t := &netlink.Iptun{LinkAttrs: netlink.LinkAttrs{Name: Tunnel, MTU: mtu}, Local: local.WeaveIP, PMtuDisc: 1} + if err := netlink.LinkAdd(t); err != nil { + return fmt.Errorf("creating egress tunnel: %v", err) + } + if link, err = netlink.LinkByName(Tunnel); err != nil { + return err + } + } + // the holder receives members' packets from its own address; only on this device + for k, v := range map[string]string{"accept_local": "1", "rp_filter": "2"} { + if err := os.WriteFile("/proc/sys/net/ipv4/conf/"+Tunnel+"/"+k, []byte(v), 0o644); err != nil { + return fmt.Errorf("sysctl %s on %s: %v", k, Tunnel, err) + } + } + return netlink.LinkSetUp(link) +} diff --git a/lb/egress_test.go b/lb/egress_test.go new file mode 100644 index 00000000..3343932b --- /dev/null +++ b/lb/egress_test.go @@ -0,0 +1,296 @@ +package lb + +import ( + "net" + "strings" + "testing" + "time" + + "golang.org/x/sys/unix" + coreapi "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// the S1 fixtures: cp, w1 (holds 86.107.168.15), w2 + +func egressFor(name string, status EgressServiceStatus) *EgressService { + return &EgressService{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name}, + Spec: EgressServiceSpec{Emit: EmitHolder}, Status: status} +} + +func mailSvc() *coreapi.Service { + return svc("mail", "86.107.168.15", func(s *coreapi.Service) { + s.Spec.Selector = map[string]string{"app": "mail"} + }) +} + +func pod(name, node, ip string, labels map[string]string, mutate ...func(*coreapi.Pod)) *coreapi.Pod { + p := &coreapi.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name, Labels: labels, + CreationTimestamp: metav1.NewTime(time.Unix(1000, 0))}, + Spec: coreapi.PodSpec{NodeName: node}, Status: coreapi.PodStatus{PodIP: ip, Phase: coreapi.PodRunning}} + for _, m := range mutate { + m(p) + } + return p +} + +var mailLabels = map[string]string{"app": "mail"} + +// the status the holder (w1) writes for three nodes, blocks in name order +func written() EgressServiceStatus { + return EgressServiceStatus{Host: "w1", Address: "86.107.168.15", Blocks: []PortBlock{ + {Node: "cp", From: 61000, To: 61511}, {Node: "w1", From: 61512, To: 62023}, {Node: "w2", From: 62024, To: 62535}}} +} + +func TestBlocksAreStableAndCapped(t *testing.T) { + got := AllocateBlocks(nil, []string{"w2", "cp", "w1"}) + if !sameStatus(EgressServiceStatus{Blocks: got}, EgressServiceStatus{Blocks: written().Blocks}) { + t.Fatalf("fresh blocks: %+v", got) + } + // a node leaves, another joins: survivors keep their blocks, the newcomer takes the freed lowest one + got = AllocateBlocks(got, []string{"w1", "w2", "w3"}) + want := []PortBlock{{"w1", 61512, 62023}, {"w2", 62024, 62535}, {"w3", 61000, 61511}} + if !sameStatus(EgressServiceStatus{Blocks: got}, EgressServiceStatus{Blocks: want}) { + t.Fatalf("after churn: %+v", got) + } + // misaligned or duplicate blocks are not kept + got = AllocateBlocks([]PortBlock{{"w1", 61001, 61512}, {"w2", 61000, 61511}, {"w3", 61000, 61511}}, []string{"w1", "w2", "w3"}) + for _, b := range got { + if (b.From-BlockBase)%BlockSize != 0 || b.To != b.From+BlockSize-1 { + t.Fatalf("invalid block kept: %+v", got) + } + } + if got[1].Node != "w2" || got[1].From != 61000 { + t.Fatalf("first valid claim wins: %+v", got) + } + var many []string + for i := 0; i < BlockCount+2; i++ { + many = append(many, string(rune('a'+i))) + } + if n := len(AllocateBlocks(nil, many)); n != BlockCount { + t.Fatalf("more blocks than exist: %d", n) + } + for _, b := range AllocateBlocks(nil, many) { + if b.To > 65535 || b.From < 61000 { + t.Fatalf("block outside the range: %+v", b) + } + } +} + +func TestHolderWritesStatusAndMembersWaitForIt(t *testing.T) { + e := egressFor("mail", EgressServiceStatus{}) + pods := []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)} + d := ComputeEgress("w1", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{e}, nil, nil) + if len(d.Status) != 1 || !sameStatus(d.Status[0].Status, written()) { + t.Fatalf("holder's status: %+v", d.Status) + } + // until the status is in the API, a member's outside traffic goes nowhere (fail closed) + d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{e}, pods, nil) + if len(d.Status) != 0 || len(d.Members) != 1 || !d.Members[0].Unreachable { + t.Fatalf("member before status: %+v", d) + } + if d.Tunnelled("w2") { + t.Fatalf("no tunnel for an unreachable member") + } +} + +func TestMemberTunnelsToHolderWhichRoutesItsBlockBack(t *testing.T) { + e := egressFor("mail", written()) + svcs := []*coreapi.Service{mailSvc()} + pods := []*coreapi.Pod{ + pod("mail-0", "w2", "10.44.0.9", mailLabels), + pod("web", "w2", "10.44.0.10", map[string]string{"app": "web"}), // not a member (A4) + } + d := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, pods, nil) + if len(d.Members) != 1 || d.Members[0].Pod != "gitea/mail-0" { + t.Fatalf("members: %+v", d.Members) + } + m := d.Members[0] + if m.Unreachable || !m.Address.Equal(net.ParseIP("86.107.168.15")) || m.From != 62024 || m.To != 62535 || m.Holder.Name != "w1" { + t.Fatalf("member path: %+v", m) + } + if !d.Tunnelled("w2") || len(d.Peers("w2")) != 1 || d.Peers("w2")[0].Name != "w1" { + t.Fatalf("w2 tunnels to w1 only: %+v", d.Peers("w2")) + } + if len(d.Status) != 0 || len(d.Routes) != 0 { + t.Fatalf("only the holder writes and routes: %+v", d) + } + // the holder: blocks of the other nodes route back to them; its own block is plain conntrack + h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, nil, nil) + if len(h.Status) != 0 { + t.Fatalf("status already right: %+v", h.Status) + } + if len(h.Routes) != 2 || h.Routes[0].Peer.Name != "cp" || h.Routes[1].Peer.Name != "w2" || h.Routes[1].From != 62024 { + t.Fatalf("holder routes: %+v", h.Routes) + } + // a member on the holder leaves from there + hm := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-1", "w1", "10.40.0.4", mailLabels)}, nil) + if len(hm.Members) != 1 || hm.Members[0].Holder.Name != "w1" || hm.Members[0].From != 61512 { + t.Fatalf("member on the holder: %+v", hm.Members) + } +} + +func TestEgressFailsClosed(t *testing.T) { + pods := []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)} + // no node holds the address: status cleared (by the first node by name), member dropped + unheld := svc("mail", "86.107.168.99", func(s *coreapi.Service) { s.Spec.Selector = mailLabels }) + for _, self := range []string{"cp", "w2"} { + d := ComputeEgress(self, nodes, []*coreapi.Service{unheld}, []*EgressService{egressFor("mail", written())}, pods, nil) + clears := len(d.Status) == 1 && d.Status[0].Status.Host == "" + if clears != (self == "cp") { + t.Fatalf("%s clears=%v: %+v", self, clears, d.Status) + } + if self == "w2" && (len(d.Members) != 1 || !d.Members[0].Unreachable) { + t.Fatalf("member of an unheld address: %+v", d.Members) + } + } + // two EgressServices select the pod: dropped, reported + other := svc("mail2", "86.107.168.15", func(s *coreapi.Service) { s.Spec.Selector = mailLabels }) + d := ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc(), other}, + []*EgressService{egressFor("mail", written()), egressFor("mail2", written())}, pods, nil) + if len(d.Members) != 1 || !d.Members[0].Unreachable || !strings.Contains(strings.Join(d.Problems, "\n"), "several EgressServices") { + t.Fatalf("double member: %+v", d) + } + // a spec this agent does not implement is refused, not guessed at + bad := egressFor("mail", written()) + bad.Spec.Emit = "Direct" + d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{bad}, pods, nil) + if len(d.Members) != 1 || !d.Members[0].Unreachable { + t.Fatalf("unknown emit mode: %+v", d) + } + // hostNetwork pods are never members (their traffic is the node's) + hn := pod("mail-h", "w2", "86.107.168.88", mailLabels, func(p *coreapi.Pod) { p.Spec.HostNetwork = true }) + d = ComputeEgress("w2", nodes, []*coreapi.Service{mailSvc()}, []*EgressService{egressFor("mail", written())}, []*coreapi.Pod{hn}, nil) + if len(d.Members) != 0 { + t.Fatalf("hostNetwork pod became a member: %+v", d.Members) + } +} + +func TestAdmissionDecidesWhoOwnsAnIP(t *testing.T) { + e := egressFor("mail", written()) + svcs := []*coreapi.Service{mailSvc()} + old := pod("mail-0", "w2", "10.44.0.9", mailLabels, func(p *coreapi.Pod) { + now := metav1.Now() + p.DeletionTimestamp = &now + }) + newer := pod("web", "w2", "", map[string]string{"app": "web"}) // API has no IP for it yet + // the CNI plugin just gave 10.44.0.9 to web: the terminating member must not keep it (A4) + d := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{old, newer}, + map[string]Admission{"10.44.0.9": {Namespace: "gitea", Name: "web"}}) + if len(d.Members) != 0 { + t.Fatalf("reused IP kept the old member's rules: %+v", d.Members) + } + // and the other way round: a member admitted before the API shows its IP is protected at once (A1) + fresh := pod("mail-1", "w2", "", mailLabels) + d = ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{fresh}, + map[string]Admission{"10.44.0.11": {Namespace: "gitea", Name: "mail-1"}}) + if len(d.Members) != 1 || d.Members[0].IP.String() != "10.44.0.11" || d.Members[0].Unreachable { + t.Fatalf("admitted member: %+v", d.Members) + } + // without an admission, a live pod wins a shared IP over a terminating one + live := pod("web", "w2", "10.44.0.9", map[string]string{"app": "web"}) + d = ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{old, live}, nil) + if len(d.Members) != 0 { + t.Fatalf("terminating member won the IP: %+v", d.Members) + } +} + +func TestRenderedEgressRules(t *testing.T) { + _, podNet, _ := net.ParseCIDR("10.32.0.0/12") + e := egressFor("mail", written()) + svcs := []*coreapi.Service{mailSvc()} + member := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil) + r := Render(Desired{}, member, "w2", podNet, nil) + for _, want := range []string{ + "-A REWEAVE-LB-CLASS -d 10.32.0.0/12 -j RETURN\n", + "-A REWEAVE-LB-CLASS -d 86.107.168.24/32 -j RETURN\n", // a node is not outside + "-A REWEAVE-LB-CLASS -s 10.44.0.9/32 -m comment --comment \"gitea/mail-0\" -j CONNMARK --set-xmark 0x100000/0x100000\n", + "-A REWEAVE-LB-CLASS -s 10.44.0.9/32 -j MARK --set-xmark 0x100000/0x100000\n", + "-A REWEAVE-LB-POST -s 10.44.0.9/32 -o reweave-eg -m mark --mark 0x100000/0x100000 -p tcp -j SNAT --to-source 86.107.168.15:62024-62535\n", + "-A REWEAVE-LB-POST -s 10.44.0.9/32 -o reweave-eg -m mark --mark 0x100000/0x100000 -p udp -j SNAT --to-source 86.107.168.15:62024-62535\n", + } { + if !strings.Contains(r.Nat, want) { + t.Fatalf("member nat lacks %q:\n%s", want, r.Nat) + } + } + for _, want := range []string{ + "-A REWEAVE-LB-FWD -s 10.44.0.9/32 -o reweave-eg -m mark --mark 0x100000/0x100000 -p tcp -j RETURN\n", + "-A REWEAVE-LB-FWD -s 10.44.0.9/32 -m mark --mark 0x100000/0x100000 -m comment --comment \"gitea/mail-0\" -j DROP\n", + "-A REWEAVE-LB-FWD -i reweave-eg -d 86.107.168.15/32 -j DROP\n", + "-A REWEAVE-LB-FWD -i reweave-eg -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN\n", + "-A REWEAVE-LB-FWD -i reweave-eg -j DROP\n", + "-A REWEAVE-LB-IN -p 4 -s 10.40.0.0/32 -m mac --mac-source aa:aa:aa:aa:aa:02 -j RETURN\n", + "-A REWEAVE-LB-IN -p 4 -j DROP\n", + } { + if !strings.Contains(r.Filter, want) { + t.Fatalf("member filter lacks %q:\n%s", want, r.Filter) + } + } + for _, want := range []string{ + "-A REWEAVE-LB-MARK -m conntrack --ctdir ORIGINAL -m connmark --mark 0x100000/0x100000 -j MARK --set-xmark 0x100000/0x100000\n", + "-A REWEAVE-LB-MSS -o reweave-eg -p tcp -m tcp --tcp-flags SYN,RST SYN -j TCPMSS --clamp-mss-to-pmtu\n", + } { + if !strings.Contains(r.Mangle, want) { + t.Fatalf("member mangle lacks %q:\n%s", want, r.Mangle) + } + } + // the guard's DROP comes after the member's own RETURNs: nothing else of its leaves + if strings.Index(r.Filter, "-p tcp -j RETURN") > strings.Index(r.Filter, "\"gitea/mail-0\" -j DROP") { + t.Fatalf("guard order:\n%s", r.Filter) + } + // the holder: accepts each node's block from the tunnel, drops the rest; a member on it SNATs to its own block + h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-1", "w1", "10.40.0.4", mailLabels)}, nil) + r = Render(Desired{}, h, "w1", podNet, nil) + for _, want := range []string{ + "-A REWEAVE-LB-FWD -i reweave-eg -s 86.107.168.15/32 -p tcp -m tcp --sport 62024:62535 -m comment --comment \"block of w2\" -j RETURN\n", + "-A REWEAVE-LB-FWD -i reweave-eg -s 86.107.168.15/32 -p udp -m udp --sport 61000:61511 -m comment --comment \"block of cp\" -j RETURN\n", + "-A REWEAVE-LB-FWD -s 10.40.0.4/32 -o weave -m mark --mark 0x100000/0x100000 -j DROP\n", + } { + if !strings.Contains(r.Filter, want) { + t.Fatalf("holder filter lacks %q:\n%s", want, r.Filter) + } + } + if !strings.Contains(r.Nat, "-A REWEAVE-LB-POST -s 10.40.0.4/32 ! -o weave -m mark --mark 0x100000/0x100000 -p tcp -j SNAT --to-source 86.107.168.15:61512-62023\n") { + t.Fatalf("holder's own member:\n%s", r.Nat) + } + for _, line := range strings.Split(r.Nat+r.Filter, "\n") { + if strings.Contains(line, "10.44.0.10") { + t.Fatalf("a rule names a non-member: %s", line) + } + } +} + +func TestPolicyRules(t *testing.T) { + e := egressFor("mail", written()) + svcs := []*coreapi.Service{mailSvc()} + h := ComputeEgress("w1", nodes, svcs, []*EgressService{e}, nil, nil) + rules := wantRules(h, "w1") + tables := peerTables(h, "w1") + if len(rules) != 4 { + t.Fatalf("2 peers x tcp,udp: %+v", rules) + } + for _, r := range rules { + if r.priority != HolderRulePriority || r.dst != "86.107.168.15/32" || r.src != "" || r.mark != 0 { + t.Fatalf("holder rule: %+v", r) + } + if r.ipproto != unix.IPPROTO_TCP && r.ipproto != unix.IPPROTO_UDP { + t.Fatalf("protocol: %+v", r) + } + if (r.from == 62024) != (r.table == tables["w2"]) || (r.from == 61000) != (r.table == tables["cp"]) { + t.Fatalf("block routed to the wrong node: %+v %v", r, tables) + } + } + if HolderRulePriority >= LocalRulePriority { + t.Fatalf("holder rules must precede the local table") + } + m := ComputeEgress("w2", nodes, svcs, []*EgressService{e}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil) + rules = wantRules(m, "w2") + if len(rules) != 1 || rules[0].src != "10.44.0.9/32" || rules[0].mark != EgressMark || rules[0].table != peerTables(m, "w2")["w1"] { + t.Fatalf("member rule: %+v", rules) + } + u := ComputeEgress("w2", nodes, svcs, []*EgressService{egressFor("mail", EgressServiceStatus{})}, []*coreapi.Pod{pod("mail-0", "w2", "10.44.0.9", mailLabels)}, nil) + rules = wantRules(u, "w2") + if len(rules) != 1 || rules[0].table != EgressTableBase { + t.Fatalf("unreachable member must hit the unreachable table: %+v", rules) + } +} diff --git a/lb/model_test.go b/lb/model_test.go index bfb17276..0455b96a 100644 --- a/lb/model_test.go +++ b/lb/model_test.go @@ -204,9 +204,9 @@ func TestRenderedRules(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.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"}) + r := Render(d, EgressDesired{}, "w1", podNet, []string{"REWEAVE-LBS-DEADBEEF0000"}) for _, want := range []string{ - "*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n", + "*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n:REWEAVE-LB-CLASS - [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", @@ -230,11 +230,11 @@ func TestRenderedRules(t *testing.T) { 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" { + if r.Mangle != "*mangle\n:REWEAVE-LB-MARK - [0:0]\n:REWEAVE-LB-MSS - [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) + r = Render(d, EgressDesired{}, "w2", 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", @@ -244,7 +244,10 @@ func TestRenderedRules(t *testing.T) { 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" { + if r.Nat != "*nat\n:REWEAVE-LB - [0:0]\n:REWEAVE-LB-POST - [0:0]\n:REWEAVE-LB-CLASS - [0:0]\nCOMMIT\n" { t.Fatalf("a non-holder forwards nothing:\n%s", r.Nat) } + if r.Filter != "*filter\n:REWEAVE-LB-FWD - [0:0]\n:REWEAVE-LB-IN - [0:0]\nCOMMIT\n" { + t.Fatalf("no EgressService, no filter rules:\n%s", r.Filter) + } } diff --git a/plugin/net/cni.go b/plugin/net/cni.go index bed6669b..3f4bc3a7 100644 --- a/plugin/net/cni.go +++ b/plugin/net/cni.go @@ -1,6 +1,7 @@ package plugin import ( + "context" "crypto/rand" "encoding/json" "errors" @@ -14,6 +15,7 @@ import ( current "github.com/containernetworking/cni/pkg/types/100" "github.com/containernetworking/plugins/pkg/ipam" weaveapi "github.com/rajch/weave/api" + "github.com/rajch/weave/lb/admit" weavenet "github.com/rajch/weave/net" ipamplugin "github.com/rajch/weave/plugin/ipam" "github.com/vishvananda/netlink" @@ -86,6 +88,18 @@ func (c *CNIPlugin) CmdAdd(args *skel.CmdArgs) error { // Only expecting one address ip := result.IPs[0] + // An EgressService member's first packet must already leave from its Service's address (or not at all): + // weave-npc installs the pod's egress rules before the pod has an interface. No answer, no pod (the + // runtime calls DEL, which releases the address, and kubelet retries). + if conf.EgressAdmission { + if ns, name := admit.PodFromArgs(args.Args); ns != "" && name != "" { + if _, err := admit.Admit(context.Background(), admit.SocketPath, + admit.Request{Namespace: ns, Name: name, IP: ip.Address.IP.String()}); err != nil { + return err + } + } + } + // If config says nothing about routes or gateway, default one will be via the bridge if len(result.Routes) == 0 && ip.Gateway == nil { bridgeIP, err := weavenet.FindBridgeIP(conf.BrName, &ip.Address) @@ -241,4 +255,6 @@ type NetConf struct { IPMasq bool `json:"ipMasq"` MTU int `json:"mtu"` HairpinMode bool `json:"hairpinMode"` + // EgressAdmission: weave-npc must admit every pod before it gets its interface (EgressService members) + EgressAdmission bool `json:"egressAdmission"` } diff --git a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml index 6c8c873c..83b3f686 100644 --- a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml +++ b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml @@ -1,6 +1,68 @@ apiVersion: v1 kind: List items: + # EgressService (reweave.levk.dev/v1alpha1), served by weave-npc's lb agent: a LoadBalancer Service's pods + # leave the cluster from the Service's address (named after the Service, same namespace) + - apiVersion: apiextensions.k8s.io/v1 + kind: CustomResourceDefinition + metadata: + name: egressservices.reweave.levk.dev + labels: + name: weave-net + spec: + group: reweave.levk.dev + scope: Namespaced + names: + plural: egressservices + singular: egressservice + kind: EgressService + listKind: EgressServiceList + versions: + - name: v1alpha1 + served: true + storage: true + subresources: + status: {} + additionalPrinterColumns: + - {name: Emit, type: string, jsonPath: .spec.emit} + - {name: Host, type: string, jsonPath: .status.host} + - {name: Address, type: string, jsonPath: .status.address} + - {name: Age, type: date, jsonPath: .metadata.creationTimestamp} + schema: + openAPIV3Schema: + type: object + description: >- + Makes every pod the same-named LoadBalancer Service (class reweave.levk.dev/node-address) + selects leave the cluster from that Service's address, or not at all. + required: [spec] + properties: + spec: + type: object + required: [emit] + properties: + sourceIPBy: + type: string + description: LoadBalancerIP - members leave from the Service's load-balancer address. + enum: [LoadBalancerIP] + default: LoadBalancerIP + emit: + type: string + description: Holder - members' packets are tunnelled to the node holding the address, which emits them. + enum: [Holder] + status: + type: object + properties: + host: {type: string} + address: {type: string} + blocks: + type: array + items: + type: object + required: [node, from, to] + properties: + node: {type: string} + from: {type: integer, format: int32} + to: {type: integer, format: int32} - apiVersion: v1 kind: ServiceAccount metadata: @@ -78,6 +140,20 @@ items: - nodes verbs: - patch + - apiGroups: + - reweave.levk.dev + resources: + - egressservices + verbs: + - get + - list + - watch + - apiGroups: + - reweave.levk.dev + resources: + - egressservices/status + verbs: + - patch - apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: @@ -226,6 +302,8 @@ items: - name: xtables-lock mountPath: /run/xtables.lock readOnly: false + - name: weave-npc-run + mountPath: /run/weave-npc hostNetwork: true dnsPolicy: ClusterFirstWithHostNet hostPID: false @@ -264,6 +342,10 @@ items: hostPath: path: /run/xtables.lock type: FileOrCreate + - name: weave-npc-run + hostPath: + path: /run/weave-npc + type: DirectoryOrCreate priorityClassName: system-node-critical updateStrategy: type: RollingUpdate diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index babb0ff0..a7b2b369 100644 --- a/prog/weave-npc/main.go +++ b/prog/weave-npc/main.go @@ -340,8 +340,11 @@ func root(cmd *cobra.Command, args []string) { 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()) + // LoadBalancer Services of class lb.Class: addresses this node holds, client address kept; and + // EgressServices: those Services' pods leave the cluster from that address (or not at all) + egress, err := lb.NewEgressClient(config) + handleFatal(err) + go lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)).Run(context.Background()) } signals := make(chan os.Signal, 1) diff --git a/reweave/test/egress/egress-test.yaml b/reweave/test/egress/egress-test.yaml new file mode 100644 index 00000000..cbaa63a3 --- /dev/null +++ b/reweave/test/egress/egress-test.yaml @@ -0,0 +1,70 @@ +# End-to-end test of EgressService (weave-npc's lb agent). Placeholders are filled by run.sh: ADDR (an +# address HOLDER holds), HOLDER, REMOTE and OTHER (the cluster's other two nodes). +apiVersion: v1 +kind: List +items: + - apiVersion: v1 + kind: Namespace + metadata: + name: egtest + - apiVersion: v1 + kind: Service + metadata: + name: mail + namespace: egtest + spec: + type: LoadBalancer + loadBalancerClass: reweave.levk.dev/node-address + loadBalancerIP: ADDR + allocateLoadBalancerNodePorts: false + selector: {egress: member} + ports: [{name: probe, port: 18090, targetPort: 8080}] + - apiVersion: reweave.levk.dev/v1alpha1 + kind: EgressService + metadata: + name: mail + namespace: egtest + spec: + emit: Holder + # members on every node, and one that is not a member + - apiVersion: v1 + kind: Pod + metadata: {name: member-remote, namespace: egtest, labels: {egress: member}} + spec: + nodeName: REMOTE + containers: [{name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}] + - apiVersion: v1 + kind: Pod + metadata: {name: member-other, namespace: egtest, labels: {egress: member}} + spec: + nodeName: OTHER + containers: [{name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}] + - apiVersion: v1 + kind: Pod + metadata: {name: member-holder, namespace: egtest, labels: {egress: member}} + spec: + nodeName: HOLDER + containers: [{name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}] + - apiVersion: v1 + kind: Pod + metadata: {name: member-dns, namespace: egtest, labels: {egress: member}} + spec: + nodeName: REMOTE + containers: [{name: c, image: "registry.k8s.io/e2e-test-images/jessie-dnsutils:1.7", command: ["sleep", "3600"]}] + - apiVersion: v1 + kind: Pod + metadata: {name: plain, namespace: egtest, labels: {egress: no}} + spec: + nodeName: REMOTE + containers: [{name: c, image: "curlimages/curl:8.10.1", command: ["sleep", "3600"]}] + # east-west: /clientip answers with the address the connection came from + - apiVersion: v1 + kind: Pod + metadata: {name: echo, namespace: egtest, labels: {app: echo}} + spec: + nodeName: OTHER + containers: + - name: echo + image: registry.k8s.io/e2e-test-images/agnhost:2.53 + args: ["netexec", "--http-port=8080"] + readinessProbe: {httpGet: {path: /healthz, port: 8080}} diff --git a/reweave/test/egress/run.sh b/reweave/test/egress/run.sh new file mode 100755 index 00000000..81c567e7 --- /dev/null +++ b/reweave/test/egress/run.sh @@ -0,0 +1,103 @@ +#!/usr/bin/env bash +# End-to-end test of EgressService against a live cluster (assertions A1-A6 of the design, plus the first +# packet). Members ask outside services which address they come from: api.ipify.org (TCP) and OpenDNS's +# myip.opendns.com (UDP). +# +# CP="ssh ... root@" HOLDER_SSH="ssh ... root@" REMOTE_SSH="ssh ... root@" \ +# ADDR=
HOLDER= REMOTE= OTHER= \ +# reweave/test/egress/run.sh +# +# Asserts, in order: +# 0. the holder writes the status: host, address, a port block for every node +# 1. the first packet of a member started after the EgressService leaves as ADDR (CNI admission) +# A1/A2. members on every node leave as ADDR (TCP and UDP) and get their replies +# A4. a non-member pod on the same node as a member leaves as its node (untouched), and no rule names it +# A3. a member talking to a pod sees its own pod address there (east-west untouched) +# A5. naming an address nobody holds clears the status, and members' outside traffic is dropped, not leaked +# A6. deleting the EgressService removes its state: members leave as their node again, rules are gone +# Leaves nothing behind (namespace egtest is deleted, also on failure). +set -euo pipefail + +: "${CP:?}" "${HOLDER_SSH:?}" "${REMOTE_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 egtest --ignore-not-found --wait=false >/dev/null 2>&1 || true; } +trap cleanup EXIT + +eventually() { + local n=$1; shift + for _ in $(seq "$n"); do + if "$@" >/dev/null 2>&1; then return 0; fi + sleep 1 + done + return 1 +} +whoami_tcp() { k -n egtest exec "$1" -- curl -s --max-time 10 https://api.ipify.org; } +# each check re-reads on every try of eventually +tcp_is() { [ "$(whoami_tcp "$1" || true)" = "$2" ]; } +whoami_udp() { k -n egtest exec "$1" -- dig +short +time=5 +tries=1 myip.opendns.com @208.67.222.222; } +node_ip() { k get node "$1" -o "jsonpath='{.status.addresses[?(@.type==\"InternalIP\")].address}'"; } + +remote_ip=$(node_ip "$REMOTE") +[ -n "$remote_ip" ] || fail "no address for $REMOTE" + +sed -e "s/\bADDR\b/$ADDR/g; s/\bHOLDER\b/$HOLDER/g; s/\bREMOTE\b/$REMOTE/g; s/\bOTHER\b/$OTHER/g" \ + "$here/egress-test.yaml" | $CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl apply -f -" >/dev/null +k -n egtest wait --for=condition=Ready pod --all --timeout=300s >/dev/null || fail "test pods not ready" + +status() { k -n egtest get egressservice mail -o "jsonpath='{.status.host} {.status.address} {.status.blocks[*].node}'"; } +status_is() { [ "$(status)" = "$1" ]; } +first_line() { k -n egtest logs member-first 2>/dev/null | head -1; } +said_something() { [ -n "$(first_line)" ]; } +want="$HOLDER $ADDR $(printf '%s\n' "$HOLDER" "$REMOTE" "$OTHER" | sort | tr '\n' ' ' | sed 's/ $//')" +eventually 60 status_is "$want" || fail "status: '$(status)', want '$want'" +ok "0. status: host $HOLDER, address $ADDR, a port block for each node" + +# a member created now: its very first connection must already leave as ADDR +$CP "KUBECONFIG=/etc/kubernetes/admin.conf kubectl -n egtest run member-first --labels egress=member --restart=Never \ + --overrides='{\"spec\":{\"nodeName\":\"$REMOTE\"}}' --image=curlimages/curl:8.10.1 --command -- \ + sh -c 'curl -s --max-time 10 https://api.ipify.org; echo; sleep 3600'" >/dev/null +eventually 120 said_something || fail "member-first said nothing" +first=$(first_line) +[ "$first" = "$ADDR" ] || fail "member-first's first connection left as '$first', want $ADDR" +ok "1. a new member's first connection leaves as $ADDR" + +for p in member-remote member-other member-holder; do + got=$(whoami_tcp $p || true) + [ "$got" = "$ADDR" ] || fail "A1: $p left as '$got', want $ADDR" +done +got=$(whoami_udp member-dns || true) +[ "$got" = "$ADDR" ] || fail "A1: member-dns (UDP) left as '$got', want $ADDR" +ok "A1/A2. members on $REMOTE, $OTHER and $HOLDER leave as $ADDR over TCP and UDP, replies come back" + +got=$(whoami_tcp plain || true) +[ "$got" = "$remote_ip" ] || fail "A4: non-member left as '$got', want its node $remote_ip" +plain_ip=$(k -n egtest get pod plain -o "jsonpath='{.status.podIP}'") +if $REMOTE_SSH "iptables-save | grep -F -- '$plain_ip/32'; ip rule | grep -F -- '$plain_ip'"; then + fail "A4: a rule names the non-member $plain_ip" +fi +ok "A4. the non-member leaves as its node ($remote_ip); no rule names it" + +echo_ip=$(k -n egtest get pod echo -o "jsonpath='{.status.podIP}'") +member_ip=$(k -n egtest get pod member-remote -o "jsonpath='{.status.podIP}'") +got=$(k -n egtest exec member-remote -- curl -s --max-time 5 "http://$echo_ip:8080/clientip" || true) +[ "${got%:*}" = "$member_ip" ] || fail "A3: echo saw '$got', want the member's pod address $member_ip" +ok "A3. east-west: the echo pod sees the member's own address ($member_ip)" + +k -n egtest patch svc mail --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"192.0.2.1\"}}'" >/dev/null +eventually 60 status_is " " || fail "A5: status not cleared: '$(status)'" +got=$(whoami_tcp member-remote || true) +[ -z "$got" ] || fail "A5: with no holder a member still got out, as '$got'" +k -n egtest patch svc mail --type merge -p "'{\"spec\":{\"loadBalancerIP\":\"$ADDR\"}}'" >/dev/null +eventually 60 status_is "$want" || fail "status not restored: '$(status)'" +eventually 30 tcp_is member-remote "$ADDR" || fail "member did not get back out as $ADDR" +ok "A5. no holder: status cleared, members dropped (not leaked); restored afterwards" + +k -n egtest delete egressservice mail >/dev/null +eventually 60 tcp_is member-remote "$remote_ip" || fail "A6: after delete, member left as '$(whoami_tcp member-remote || true)'" +if $REMOTE_SSH "ip rule | grep -E '^9100:'"; then fail "A6: member rules left on $REMOTE"; fi +if $HOLDER_SSH "ip rule | grep -E '^90:'"; then fail "A6: holder rules left on $HOLDER"; fi +ok "A6. EgressService deleted: members leave as their node again, no rules left" +echo PASS diff --git a/reweave/test/lb/run.sh b/reweave/test/lb/run.sh index d9383947..ad3f33c4 100755 --- a/reweave/test/lb/run.sh +++ b/reweave/test/lb/run.sh @@ -42,9 +42,12 @@ sed -e "s/\bADDR\b/$ADDR/g; s/\bHOLDER\b/$HOLDER/g; s/\bREMOTE\b/$REMOTE/g; s/\b 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}'"; } +# each check re-reads on every try of eventually +status_is() { [ "$(status "$1")" = "$2" ]; } +chains_gone() { [ "$($HOLDER_SSH "iptables-save -t nat | grep -c '^:REWEAVE-LBS-'" || true)" = 0 ]; } 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" + eventually 30 status_is $s "$want" || fail "$s status: $(status $s), want $want" done ok "1. status [{ip: $ADDR, ipMode: Proxy}] on both Services" @@ -70,14 +73,13 @@ k -n lbtest exec client -- /agnhost connect "$ADDR:18080" --timeout 5s >/dev/nul 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)" +eventually 30 status_is 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" +eventually 30 chains_gone || fail "holder kept per-port chains after the Services went away" ok "6. Services deleted: holder's chains removed" echo PASS diff --git a/weave b/weave index 0eea26b8..ee134620 100755 --- a/weave +++ b/weave @@ -763,7 +763,8 @@ create_cni_config() { { "name": "weave", "type": "weave-net", - "hairpinMode": $2 + "hairpinMode": $2, + "egressAdmission": $3 }, { "type": "portmap", @@ -778,6 +779,9 @@ EOF setup_cni() { # if env var HAIRPIN_MODE is not set, default it to true HAIRPIN_MODE=${HAIRPIN_MODE:-true} + # weave-npc admits every pod before it gets its interface (EgressService members leave from their + # Service's address from the first packet); set to false only where weave-npc runs with --lb=false + EGRESS_ADMISSION=${EGRESS_ADMISSION:-true} if install_cni_plugin $CNI_PLUGIN_DIR ; then upgrade_cni_plugin $CNI_PLUGIN_DIR fi @@ -786,8 +790,15 @@ setup_cni() { # clusters with Weave version <= 2.4.0 remove existing 10-weave.conf rm -f $HOST_ROOT/etc/cni/net.d/10-weave.conf - if [ -d $HOST_ROOT/etc/cni/net.d -a ! -f $HOST_ROOT/etc/cni/net.d/10-weave.conflist ] ; then - create_cni_config $HOST_ROOT/etc/cni/net.d/10-weave.conflist $HAIRPIN_MODE + # (re)write the conflist whenever what it should say changed, so upgrades reach existing nodes + if [ -d $HOST_ROOT/etc/cni/net.d ] ; then + CONFLIST=$HOST_ROOT/etc/cni/net.d/10-weave.conflist + create_cni_config $CONFLIST.new $HAIRPIN_MODE $EGRESS_ADMISSION + if [ -f $CONFLIST ] && [ "$(cat $CONFLIST.new)" = "$(cat $CONFLIST)" ] ; then + rm -f $CONFLIST.new + else + mv -f $CONFLIST.new $CONFLIST + fi fi }