levaliciousandClaude Opus 5.5
ef50dc921f
images / images (push) Successful in 14m35s
iptables refuses a DNAT to ip:port unless the same rule names the
protocol, so every per-port chain's DNAT now has -p. Live on the SG
cluster the first sync (no endpoints yet) applied; every later one failed
with "Need TCP, UDP, SCTP or DCCP with port specification" and left the
chains empty. The unit test now rejects any DNAT line without its
protocol; the rendered nat and mangle payloads were also checked with the
nodes' own iptables-restore --test (nf_tables).
reweave/test/lb/run.sh is the end-to-end test: from outside, both a pod on
the holder and one on another node see the client's own address; status
is [{ip, ipMode: Proxy}] and kube-proxy has nothing for the address; a pod
on a third node reaches it; an unheld address clears the status and stops
the forward; deleting the Services removes the holder's chains.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ
248 lines
10 KiB
Go
248 lines
10 KiB
Go
package lb
|
|
|
|
import (
|
|
"fmt"
|
|
"net"
|
|
"strings"
|
|
"testing"
|
|
|
|
coreapi "k8s.io/api/core/v1"
|
|
discovery "k8s.io/api/discovery/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
)
|
|
|
|
func mac(s string) net.HardwareAddr {
|
|
m, err := net.ParseMAC(s)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return m
|
|
}
|
|
|
|
var (
|
|
cp = Node{Name: "cp", Addresses: []net.IP{net.ParseIP("86.107.168.24").To4()},
|
|
WeaveIP: net.ParseIP("10.32.0.1").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:01")}
|
|
w1 = Node{Name: "w1", Addresses: []net.IP{net.ParseIP("86.107.168.15").To4()},
|
|
WeaveIP: net.ParseIP("10.40.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:02")}
|
|
w2 = Node{Name: "w2", Addresses: []net.IP{net.ParseIP("86.107.168.88").To4()},
|
|
WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:03")}
|
|
nodes = []Node{w2, cp, w1}
|
|
)
|
|
|
|
func svc(name, ip string, mutate ...func(*coreapi.Service)) *coreapi.Service {
|
|
class := Class
|
|
s := &coreapi.Service{
|
|
ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: name},
|
|
Spec: coreapi.ServiceSpec{
|
|
Type: coreapi.ServiceTypeLoadBalancer,
|
|
LoadBalancerClass: &class,
|
|
LoadBalancerIP: ip,
|
|
Ports: []coreapi.ServicePort{
|
|
{Name: "ssh", Protocol: coreapi.ProtocolTCP, Port: 22},
|
|
{Name: "https", Port: 443}, // protocol defaults to TCP
|
|
},
|
|
},
|
|
}
|
|
for _, m := range mutate {
|
|
m(s)
|
|
}
|
|
return s
|
|
}
|
|
|
|
func shows(ip string) func(*coreapi.Service) {
|
|
return func(s *coreapi.Service) {
|
|
s.Status.LoadBalancer.Ingress = []coreapi.LoadBalancerIngress{{IP: ip}}
|
|
}
|
|
}
|
|
|
|
func ptr[T any](v T) *T { return &v }
|
|
|
|
func slice(svcName string, eps map[string]bool, ports map[string]int32) *discovery.EndpointSlice {
|
|
sl := &discovery.EndpointSlice{
|
|
ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: svcName + "-x", Labels: map[string]string{serviceNameLabel: svcName}},
|
|
AddressType: discovery.AddressTypeIPv4,
|
|
}
|
|
for ip, ready := range eps {
|
|
sl.Endpoints = append(sl.Endpoints, discovery.Endpoint{Addresses: []string{ip}, Conditions: discovery.EndpointConditions{Ready: ptr(ready)}})
|
|
}
|
|
for name, port := range ports {
|
|
sl.Ports = append(sl.Ports, discovery.EndpointPort{Name: ptr(name), Protocol: ptr(coreapi.ProtocolTCP), Port: ptr(port)})
|
|
}
|
|
return sl
|
|
}
|
|
|
|
func TestHolderForwardsToReadyEndpointsAndWritesStatus(t *testing.T) {
|
|
s := svc("gitea", "86.107.168.15")
|
|
sl := slice("gitea", map[string]bool{"10.40.0.7": true, "10.44.0.3": true, "10.44.0.9": false},
|
|
map[string]int32{"ssh": 2222, "https": 3000})
|
|
d := Compute("w1", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl})
|
|
if len(d.Problems) != 0 {
|
|
t.Fatalf("problems: %v", d.Problems)
|
|
}
|
|
if len(d.Forwards) != 1 || !d.Forwards[0].Address.Equal(net.ParseIP("86.107.168.15")) {
|
|
t.Fatalf("forwards: %+v", d.Forwards)
|
|
}
|
|
f := d.Forwards[0]
|
|
if len(f.Ports) != 2 || f.Ports[0].Port != 22 || f.Ports[1].Port != 443 || f.Ports[1].Protocol != coreapi.ProtocolTCP {
|
|
t.Fatalf("ports: %+v", f.Ports)
|
|
}
|
|
if got := f.Ports[0].Endpoints; len(got) != 2 || got[0].String() != "10.40.0.7:2222" || got[1].String() != "10.44.0.3:2222" {
|
|
t.Fatalf("ssh endpoints (not-ready one must be left out): %v", got)
|
|
}
|
|
if got := f.Ports[1].Endpoints; len(got) != 2 || got[0].Port != 3000 {
|
|
t.Fatalf("https endpoints: %v", got)
|
|
}
|
|
if len(d.Status) != 1 || !d.Status[0].Address.Equal(net.ParseIP("86.107.168.15")) {
|
|
t.Fatalf("status: %+v", d.Status)
|
|
}
|
|
if len(d.Holders) != 0 {
|
|
t.Fatalf("a holder does not steer its own replies: %+v", d.Holders)
|
|
}
|
|
// already shown: nothing to write
|
|
d = Compute("w1", nodes, []*coreapi.Service{svc("gitea", "86.107.168.15", shows("86.107.168.15"))}, nil)
|
|
if len(d.Status) != 0 {
|
|
t.Fatalf("status rewritten: %+v", d.Status)
|
|
}
|
|
}
|
|
|
|
func TestOtherNodesSteerRepliesToTheHolderAndNeverWrite(t *testing.T) {
|
|
s := svc("gitea", "86.107.168.15")
|
|
for _, self := range []string{"cp", "w2"} {
|
|
d := Compute(self, nodes, []*coreapi.Service{s}, nil)
|
|
if len(d.Forwards) != 0 || len(d.Status) != 0 {
|
|
t.Fatalf("%s: only the holder forwards and writes: %+v", self, d)
|
|
}
|
|
if len(d.Holders) != 1 || d.Holders[0].Node.Name != "w1" || d.Holders[0].Mark < 1 || d.Holders[0].Mark > MaxMarks {
|
|
t.Fatalf("%s: holders: %+v", self, d.Holders)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestUnheldOrContestedAddressFailsClosed(t *testing.T) {
|
|
// nobody holds it: no forward anywhere; the stale status is cleared by exactly one node (first by name)
|
|
s := svc("gitea", "86.107.168.99", shows("86.107.168.99"))
|
|
for _, self := range []string{"cp", "w1", "w2"} {
|
|
d := Compute(self, nodes, []*coreapi.Service{s}, nil)
|
|
if len(d.Forwards) != 0 || len(d.Holders) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "no node holds") {
|
|
t.Fatalf("%s: %+v", self, d)
|
|
}
|
|
clears := len(d.Status) == 1 && d.Status[0].Address == nil
|
|
if clears != (self == "cp") {
|
|
t.Fatalf("%s: clears=%v", self, clears)
|
|
}
|
|
}
|
|
// two nodes claim it: neither serves it, its holder's view clears the status
|
|
both := w2
|
|
both.Addresses = append([]net.IP{net.ParseIP("86.107.168.15").To4()}, w2.Addresses...)
|
|
d := Compute("w1", []Node{cp, w1, both}, []*coreapi.Service{svc("gitea", "86.107.168.15", shows("86.107.168.15"))}, nil)
|
|
if len(d.Forwards) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "several nodes") {
|
|
t.Fatalf("contested: %+v", d)
|
|
}
|
|
// a bad request is reported, not guessed at
|
|
d = Compute("w1", nodes, []*coreapi.Service{svc("gitea", "")}, nil)
|
|
if len(d.Forwards) != 0 || len(d.Problems) != 1 || !strings.Contains(d.Problems[0], "spec.loadBalancerIP") {
|
|
t.Fatalf("no address: %+v", d)
|
|
}
|
|
}
|
|
|
|
func TestOnlyOptedInLoadBalancersAreServed(t *testing.T) {
|
|
other := svc("a", "86.107.168.15", func(s *coreapi.Service) { s.Spec.LoadBalancerClass = ptr("example.com/other") })
|
|
noClass := svc("b", "86.107.168.15", func(s *coreapi.Service) { s.Spec.LoadBalancerClass = nil })
|
|
clusterIP := svc("c", "86.107.168.15", func(s *coreapi.Service) {
|
|
s.Spec.Type = coreapi.ServiceTypeClusterIP
|
|
})
|
|
d := Compute("w1", nodes, []*coreapi.Service{other, noClass, clusterIP}, nil)
|
|
if len(d.Forwards)+len(d.Status)+len(d.Holders)+len(d.Problems) != 0 {
|
|
t.Fatalf("touched a Service that did not opt in: %+v", d)
|
|
}
|
|
}
|
|
|
|
func TestParseNodeAndAnnotationsRoundTrip(t *testing.T) {
|
|
l := Local{Addresses: []net.IP{net.ParseIP("86.107.168.88").To4(), net.ParseIP("86.107.168.15").To4()},
|
|
WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mac("aa:aa:aa:aa:aa:03")}
|
|
a := l.Annotations()
|
|
if a[AnnAddresses] != "86.107.168.15,86.107.168.88" || a[AnnWeave] != "10.44.0.0,aa:aa:aa:aa:aa:03" {
|
|
t.Fatalf("annotations: %v", a)
|
|
}
|
|
n, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "w2", Annotations: a}})
|
|
if err != nil || !n.Holds(net.ParseIP("86.107.168.15")) || !n.WeaveIP.Equal(l.WeaveIP) || n.WeaveMAC.String() != "aa:aa:aa:aa:aa:03" {
|
|
t.Fatalf("parsed %+v, %v", n, err)
|
|
}
|
|
for _, bad := range []map[string]string{{AnnAddresses: "1.2.3.4,nope"}, {AnnWeave: "10.44.0.0"}, {AnnWeave: "x,aa:aa:aa:aa:aa:03"}} {
|
|
if _, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "w2", Annotations: bad}}); err == nil {
|
|
t.Fatalf("accepted %v", bad)
|
|
}
|
|
}
|
|
if n, err := ParseNode(&coreapi.Node{ObjectMeta: metav1.ObjectMeta{Name: "new"}}); err != nil || len(n.Addresses) != 0 || n.WeaveIP != nil {
|
|
t.Fatalf("a node that published nothing holds nothing: %+v %v", n, err)
|
|
}
|
|
}
|
|
|
|
func TestMarksAreDistinctAndStable(t *testing.T) {
|
|
var ns []Node
|
|
for _, name := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
|
|
ns = append(ns, Node{Name: name})
|
|
}
|
|
first := assignMarks(ns)
|
|
seen := map[uint32]bool{}
|
|
for _, h := range first {
|
|
if seen[h.Mark] || h.Mark < 1 || h.Mark > MaxMarks {
|
|
t.Fatalf("bad or repeated mark: %+v", first)
|
|
}
|
|
seen[h.Mark] = true
|
|
}
|
|
again := assignMarks(ns)
|
|
for i := range first {
|
|
if first[i].Mark != again[i].Mark {
|
|
t.Fatalf("marks moved between passes")
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRenderedRules(t *testing.T) {
|
|
_, podNet, _ := net.ParseCIDR("10.32.0.0/12")
|
|
s := svc("gitea", "86.107.168.15")
|
|
sl := slice("gitea", map[string]bool{"10.40.0.7": true, "10.44.0.3": true, "10.44.0.4": true}, map[string]int32{"ssh": 2222, "https": 3000})
|
|
d := Compute("w1", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl})
|
|
r := Render(d, podNet, []string{"REWEAVE-LBS-DEADBEEF0000"})
|
|
for _, want := range []string{
|
|
"*nat\n:REWEAVE-LB - [0:0]\n",
|
|
":REWEAVE-LBS-DEADBEEF0000 - [0:0]\n", // stale chain emptied
|
|
"-A REWEAVE-LB -d 86.107.168.15/32 -p tcp -m tcp --dport 22 -m comment --comment \"gitea/gitea\" -j REWEAVE-LBS-",
|
|
"-p tcp -m statistic --mode random --probability 0.3333333333 -j DNAT --to-destination 10.40.0.7:2222\n",
|
|
"-p tcp -m statistic --mode random --probability 0.5000000000 -j DNAT --to-destination 10.44.0.3:2222\n",
|
|
"-p tcp -j DNAT --to-destination 10.44.0.4:2222\n",
|
|
"COMMIT\n",
|
|
} {
|
|
if !strings.Contains(r.Nat, want) {
|
|
t.Fatalf("nat lacks %q:\n%s", want, r.Nat)
|
|
}
|
|
}
|
|
for _, line := range strings.Split(r.Nat, "\n") {
|
|
// iptables refuses a DNAT to ip:port without the protocol in the same rule
|
|
if strings.Contains(line, "-j DNAT") && !strings.Contains(line, " -p tcp ") {
|
|
t.Fatalf("DNAT without its protocol: %q", line)
|
|
}
|
|
}
|
|
if strings.Contains(r.Nat, "SNAT") || strings.Contains(r.Nat, "MASQUERADE") {
|
|
t.Fatalf("the client's address must survive:\n%s", r.Nat)
|
|
}
|
|
if r.Mangle != "*mangle\n:REWEAVE-LB-MARK - [0:0]\nCOMMIT\n" {
|
|
t.Fatalf("a holder marks nothing:\n%s", r.Mangle)
|
|
}
|
|
d = Compute("w2", nodes, []*coreapi.Service{s}, []*discovery.EndpointSlice{sl})
|
|
r = Render(d, podNet, nil)
|
|
m := d.Holders[0].Mark << 24
|
|
for _, want := range []string{
|
|
"-A REWEAVE-LB-MARK -m conntrack --ctdir REPLY -m connmark ! --mark 0x0/0xff000000 -j CONNMARK --restore-mark --nfmask 0xff000000 --ctmask 0xff000000\n",
|
|
"-A REWEAVE-LB-MARK -i weave -m mac --mac-source aa:aa:aa:aa:aa:02 ! -s 10.32.0.0/12 -d 10.32.0.0/12 -m conntrack --ctstate NEW -m comment --comment \"holder w1\" -j CONNMARK --set-xmark " + fmt.Sprintf("0x%x", m) + "/0xff000000\n",
|
|
} {
|
|
if !strings.Contains(r.Mangle, want) {
|
|
t.Fatalf("mangle lacks %q:\n%s", want, r.Mangle)
|
|
}
|
|
}
|
|
if r.Nat != "*nat\n:REWEAVE-LB - [0:0]\nCOMMIT\n" {
|
|
t.Fatalf("a non-holder forwards nothing:\n%s", r.Nat)
|
|
}
|
|
}
|