diff --git a/.gitignore b/.gitignore index 9671228f..bcccc7f2 100644 --- a/.gitignore +++ b/.gitignore @@ -68,6 +68,7 @@ test/tls/tls test/tls/*.pem test/coverage test/coverage.* +test/images/network-tester/webserver *qemu-*-static terraform.tfstate terraform.tfstate.backup diff --git a/circle.yml b/circle.yml index 7cb13153..e1cd0983 100644 --- a/circle.yml +++ b/circle.yml @@ -44,7 +44,7 @@ test: parallel: true - "$SRCDIR/bin/circle-test-smoke": parallel: true - timeout: 300 + timeout: 360 post: - "$SRCDIR/bin/circle-test-teardown": parallel: true diff --git a/ipam/allocator.go b/ipam/allocator.go index b17592a0..3fc5e1e9 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -74,6 +74,7 @@ type Allocator struct { isKnownPeer func(mesh.PeerName) bool quorum func() uint now func() time.Time + tracker tracker.LocalRangeTracker } // PreClaims are IP addresses discovered before we could initialize IPAM @@ -130,6 +131,7 @@ func NewAllocator(config Config) *Allocator { quorum: config.Quorum, dead: make(map[string]time.Time), now: time.Now, + tracker: config.Tracker, } alloc.pendingClaims = make([]operation, len(config.PreClaims)) diff --git a/ipam/http.go b/ipam/http.go index b024f3ae..187a6b4e 100644 --- a/ipam/http.go +++ b/ipam/http.go @@ -88,7 +88,7 @@ func (alloc *Allocator) handleHTTPClaim(dockerCli *docker.Client, w http.Respons } // HandleHTTP wires up ipams HTTP endpoints to the provided mux. -func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, tracker string, dockerCli *docker.Client) { +func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, dockerCli *docker.Client) { router.Methods("GET").Path("/ipinfo/defaultsubnet").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "%s", defaultSubnet) }) @@ -199,6 +199,6 @@ func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CID }) router.Methods("GET").Path("/ipinfo/tracker").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - fmt.Fprintf(w, tracker) + fmt.Fprintf(w, alloc.tracker.String()) }) } diff --git a/ipam/http_test.go b/ipam/http_test.go index 1849ebb8..14ba7d41 100644 --- a/ipam/http_test.go +++ b/ipam/http_test.go @@ -43,7 +43,7 @@ func listenHTTP(alloc *Allocator, subnet address.CIDR) int { router.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, fmt.Sprintln(alloc)) }) - alloc.HandleHTTP(router, subnet, "", nil) + alloc.HandleHTTP(router, subnet, nil) httpListener, err := net.Listen("tcp", ":0") if err != nil { diff --git a/ipam/tracker/awsvpc.go b/ipam/tracker/awsvpc.go index d4836953..fdd4f313 100644 --- a/ipam/tracker/awsvpc.go +++ b/ipam/tracker/awsvpc.go @@ -84,7 +84,7 @@ func NewAWSVPCTracker(bridgeName string) (*AWSVPCTracker, error) { func (t *AWSVPCTracker) HandleUpdate(prevRanges, currRanges []address.Range, local bool) error { t.debugf("replacing %q by %q; local(%t)", prevRanges, currRanges, local) - prev, curr := removeCommon(address.NewCIDRs(merge(prevRanges)), address.NewCIDRs(merge(currRanges))) + prev, curr := RemoveCommon(address.NewCIDRs(Merge(prevRanges)), address.NewCIDRs(Merge(currRanges))) // It might make sense to do the removal first and then add entries // because of the 50 routes limit. However, in such case a container might @@ -123,6 +123,10 @@ func (t *AWSVPCTracker) HandleUpdate(prevRanges, currRanges []address.Range, loc return nil } +func (t *AWSVPCTracker) String() string { + return "awsvpc" +} + func (t *AWSVPCTracker) createVPCRoute(cidr string) (*ec2.CreateRouteOutput, error) { route := &ec2.CreateRouteInput{ RouteTableId: &t.routeTableID, @@ -232,47 +236,6 @@ func (t *AWSVPCTracker) infof(fmt string, args ...interface{}) { // Helpers -// merge merges adjacent range entries. -// The given slice has to be sorted in increasing order. -func merge(r []address.Range) []address.Range { - var merged []address.Range - - for i := range r { - if prev := len(merged) - 1; prev >= 0 && merged[prev].End == r[i].Start { - merged[prev].End = r[i].End - } else { - merged = append(merged, r[i]) - } - } - - return merged -} - -// removeCommon filters out CIDR ranges which are contained in both a and b slices. -// Both slices have to be sorted in increasing order. -func removeCommon(a, b []address.CIDR) (newA, newB []address.CIDR) { - i, j := 0, 0 - - for i < len(a) && j < len(b) { - switch { - case a[i].Start() < b[j].Start() || a[i].End() < b[j].End(): - newA = append(newA, a[i]) - i++ - case a[i].Start() > b[j].Start() || a[i].End() > b[j].End(): - newB = append(newB, b[j]) - j++ - default: - i++ - j++ - } - - } - newA = append(newA, a[i:]...) - newB = append(newB, b[j:]...) - - return -} - func parseCIDR(cidr string) (*net.IPNet, error) { ip, ipnet, err := net.ParseCIDR(cidr) if err != nil { diff --git a/ipam/tracker/helpers.go b/ipam/tracker/helpers.go new file mode 100644 index 00000000..b8823007 --- /dev/null +++ b/ipam/tracker/helpers.go @@ -0,0 +1,46 @@ +package tracker + +import ( + "github.com/weaveworks/weave/net/address" +) + +// Merge merges adjacent range entries. +// The given slice has to be sorted in increasing order. +func Merge(r []address.Range) []address.Range { + var merged []address.Range + + for i := range r { + if prev := len(merged) - 1; prev >= 0 && merged[prev].End == r[i].Start { + merged[prev].End = r[i].End + } else { + merged = append(merged, r[i]) + } + } + + return merged +} + +// RemoveCommon filters out CIDR ranges which are contained in both a and b slices. +// Both slices have to be sorted in increasing order. +func RemoveCommon(a, b []address.CIDR) (newA, newB []address.CIDR) { + i, j := 0, 0 + + for i < len(a) && j < len(b) { + switch { + case a[i].Start() < b[j].Start() || a[i].End() < b[j].End(): + newA = append(newA, a[i]) + i++ + case a[i].Start() > b[j].Start() || a[i].End() > b[j].End(): + newB = append(newB, b[j]) + j++ + default: + i++ + j++ + } + + } + newA = append(newA, a[i:]...) + newB = append(newB, b[j:]...) + + return +} diff --git a/ipam/tracker/awsvpc_test.go b/ipam/tracker/helpers_test.go similarity index 95% rename from ipam/tracker/awsvpc_test.go rename to ipam/tracker/helpers_test.go index b99500a1..dce74eec 100644 --- a/ipam/tracker/awsvpc_test.go +++ b/ipam/tracker/helpers_test.go @@ -19,7 +19,7 @@ var ( func TestRemoveCommon(t *testing.T) { a := []address.CIDR{r0to127, r1dot0to255} b := []address.CIDR{r1dot0to255, r2dot0to255} - newA, newB := removeCommon(a, b) + newA, newB := RemoveCommon(a, b) require.Equal(t, []address.CIDR{r0to127}, newA) require.Equal(t, []address.CIDR{r2dot0to255}, newB) } @@ -30,7 +30,7 @@ func TestMerge(t *testing.T) { r128to255.Range(), r2dot0to255.Range(), } - require.Equal(t, []address.Range{r0to255.Range(), r2dot0to255.Range()}, merge(ranges)) + require.Equal(t, []address.Range{r0to255.Range(), r2dot0to255.Range()}, Merge(ranges)) } // Helper diff --git a/ipam/tracker/tracker.go b/ipam/tracker/tracker.go index 6a9846db..61302e2a 100644 --- a/ipam/tracker/tracker.go +++ b/ipam/tracker/tracker.go @@ -17,4 +17,7 @@ type LocalRangeTracker interface { // The local parameter indicates whether the ranges belong to the peer // by which the method is called. HandleUpdate(prevRanges, currRanges []address.Range, local bool) error + + // String returns the tracker name + String() string } diff --git a/net/bridge.go b/net/bridge.go index 504a5d12..a3b0fd41 100644 --- a/net/bridge.go +++ b/net/bridge.go @@ -10,8 +10,13 @@ import ( "github.com/pkg/errors" "github.com/sirupsen/logrus" "github.com/vishvananda/netlink" + "k8s.io/apimachinery/pkg/types" + "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/common/odp" + "github.com/weaveworks/weave/ipam/tracker" + "github.com/weaveworks/weave/net/address" + "github.com/weaveworks/weave/net/ipset" ) /* This code implements three possible configurations to connect @@ -47,11 +52,12 @@ datapath of old kernel versions (https://github.com/weaveworks/weave/issues/1577 */ const ( - WeaveBridgeName = "weave" - DatapathName = "datapath" - DatapathIfName = "vethwe-datapath" - BridgeIfName = "vethwe-bridge" - PcapIfName = "vethwe-pcap" + WeaveBridgeName = "weave" + DatapathName = "datapath" + DatapathIfName = "vethwe-datapath" + BridgeIfName = "vethwe-bridge" + PcapIfName = "vethwe-pcap" + NoMasqLocalIpset = ipset.Name("weaver-no-masq-local") ) type Bridge interface { @@ -213,6 +219,7 @@ type BridgeConfig struct { MTU int Mac string Port int + NoMasqLocal bool } func (config *BridgeConfig) configuredBridgeType() Bridge { @@ -226,7 +233,7 @@ func (config *BridgeConfig) configuredBridgeType() Bridge { } } -func EnsureBridge(procPath string, config *BridgeConfig, log *logrus.Logger) (Bridge, error) { +func EnsureBridge(procPath string, config *BridgeConfig, log *logrus.Logger, ips ipset.Interface) (Bridge, error) { existingBridgeType, err := ExistingBridgeType(config.WeaveBridgeName, config.DatapathName) if err != nil { return nil, err @@ -252,7 +259,7 @@ func EnsureBridge(procPath string, config *BridgeConfig, log *logrus.Logger) (Br break } - if err := configureIPTables(config); err != nil { + if err := configureIPTables(config, ips); err != nil { return bridgeType, errors.Wrap(err, "configuring iptables") } @@ -279,6 +286,12 @@ func EnsureBridge(procPath string, config *BridgeConfig, log *logrus.Logger) (Br return bridgeType, errors.Wrapf(err, "configuring ARP cache on bridge %q", config.WeaveBridgeName) } + // NB: No concurrent call to Expose is possible, as EnsureBridge is called + // before any service has been started. + if err := reexpose(config, log); err != nil { + return bridgeType, err + } + return bridgeType, nil } @@ -404,7 +417,7 @@ func (f fastdpImpl) attach(veth *netlink.Veth) error { return odp.AddDatapathInterfaceIfNotExist(f.datapathName, veth.Attrs().Name) } -func configureIPTables(config *BridgeConfig) error { +func configureIPTables(config *BridgeConfig, ips ipset.Interface) error { ipt, err := iptables.New() if err != nil { return errors.Wrap(err, "creating iptables object") @@ -462,8 +475,11 @@ func configureIPTables(config *BridgeConfig) error { } if !config.NPC { - // Create a chain for allowing ingress traffic when the bridge is exposed - _ = ipt.NewChain("filter", "WEAVE-EXPOSE") + // Create/Flush a chain for allowing ingress traffic when the bridge is exposed + if err := ipt.ClearChain("filter", "WEAVE-EXPOSE"); err != nil { + return errors.Wrap(err, "failed to clear/create filter/WEAVE-EXPOSE chain") + } + fwdRules = append(fwdRules, []string{"-o", config.WeaveBridgeName, "-j", "WEAVE-EXPOSE"}) } @@ -476,13 +492,74 @@ func configureIPTables(config *BridgeConfig) error { return err } - // create a chain for masquerading - // - // NB: we do not clear the chain to preserve existing rules - // inserted by "weave expose". - _ = ipt.NewChain("nat", "WEAVE") + // Create a chain for masquerading + if err := ipt.ClearChain("nat", "WEAVE"); err != nil { + return errors.Wrap(err, "failed to clear/create nat/WEAVE chain") + } + if err := ipt.AppendUnique("nat", "POSTROUTING", "-j", "WEAVE"); err != nil { + return err + } - return ipt.AppendUnique("nat", "POSTROUTING", "-j", "WEAVE") + // For the cases where the weave bridge is the default gateway for + // containers (e.g. Kubernetes): create the ipset to store CIDRs allocated + // by IPAM for local containers. In the case of Kubernetes, external traffic + // sent to these CIDRs avoids SNAT'ing so that NodePort with + // `"externalTrafficPolicy":"Local"` would receive packets with correct + // src IP addr. + if config.NoMasqLocal { + ips := ipset.New(common.LogLogger()) + _ = ips.Destroy(NoMasqLocalIpset) + if err := ips.Create(NoMasqLocalIpset, ipset.HashNet); err != nil { + return err + } + if err := ipt.Insert("nat", "WEAVE", 1, + "-m", "set", "--match-set", string(NoMasqLocalIpset), "dst", + "-m", "comment", "--comment", "Prevent SNAT to locally running containers", + "-j", "RETURN"); err != nil { + return err + } + } + + return nil +} + +type NoMasqLocalTracker struct { + ips ipset.Interface + owner types.UID +} + +func NewNoMasqLocalTracker(ips ipset.Interface) *NoMasqLocalTracker { + return &NoMasqLocalTracker{ + ips: ips, + owner: types.UID(0), // dummy ipset owner + } +} + +func (t *NoMasqLocalTracker) String() string { + return "no-masq-local" +} + +func (t *NoMasqLocalTracker) HandleUpdate(prevRanges, currRanges []address.Range, local bool) error { + if !local { + return nil + } + + prev, curr := tracker.RemoveCommon( + address.NewCIDRs(tracker.Merge(prevRanges)), + address.NewCIDRs(tracker.Merge(currRanges))) + + for _, cidr := range curr { + if err := t.ips.AddEntry(t.owner, NoMasqLocalIpset, cidr.String(), ""); err != nil { + return err + } + } + for _, cidr := range prev { + if err := t.ips.DelEntry(t.owner, NoMasqLocalIpset, cidr.String()); err != nil { + return err + } + } + + return nil } func linkSetUpByName(linkName string) error { @@ -529,3 +606,29 @@ func ensureRulesAtTop(table, chain string, rulespecs [][]string, ipt *iptables.I return nil } + +func reexpose(config *BridgeConfig, log *logrus.Logger) error { + // Get existing IP addrs of the weave bridge. + // If the bridge hasn't been exposed, then this functions does nothing. + // + // Ideally, we should consult IPAM for IP addrs allocated to "weave:expose", + // but we don't want to introduce dependency on IPAM, as weave should be able + // to run w/o IPAM. + link, err := netlink.LinkByName(config.WeaveBridgeName) + if err != nil { + return errors.Wrapf(err, "cannot find bridge %q", config.WeaveBridgeName) + } + addrs, err := netlink.AddrList(link, netlink.FAMILY_V4) + if err != nil { + return errors.Wrapf(err, "cannot list IPv4 addrs of bridge %q", config.WeaveBridgeName) + } + + for _, addr := range addrs { + log.Infof("Re-exposing %s on bridge %q", addr.IPNet, config.WeaveBridgeName) + if err := Expose(config.WeaveBridgeName, addr.IPNet, config.AWSVPC, config.NPC); err != nil { + return errors.Wrapf(err, "unable to re-expose %s on bridge: %q", addr.IPNet, config.WeaveBridgeName) + } + } + + return nil +} diff --git a/npc/ipset/ipset.go b/net/ipset/ipset.go similarity index 93% rename from npc/ipset/ipset.go rename to net/ipset/ipset.go index 1e57b1d7..37488008 100644 --- a/npc/ipset/ipset.go +++ b/net/ipset/ipset.go @@ -1,6 +1,8 @@ package ipset import ( + "crypto/rand" + "encoding/hex" "log" "os/exec" "strings" @@ -16,6 +18,7 @@ type Type string const ( ListSet = Type("list:set") HashIP = Type("hash:ip") + HashNet = Type("hash:net") ) type Interface interface { @@ -55,7 +58,14 @@ func New(logger *log.Logger) Interface { } // Check for comment support - testIpsetName := Name("weave-test-comment") + + // To prevent from a race when more than one process check for the support + // we append a random nonce to the test ipset name. The final name is + // shorter than 31 chars (max ipset name). + nonce := make([]byte, 4) + rand.Read(nonce) + testIpsetName := Name("weave-test-comment" + hex.EncodeToString(nonce)) + // Clear it out if it already exists _ = ips.Destroy(testIpsetName) // Test for comment support diff --git a/npc/analyser.go b/npc/analyser.go index e73d37b8..01cdf5ee 100644 --- a/npc/analyser.go +++ b/npc/analyser.go @@ -8,7 +8,7 @@ import ( networkingv1 "k8s.io/api/networking/v1" "k8s.io/apimachinery/pkg/util/intstr" - "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/net/ipset" ) // analysePolicyLegacy is used to analyse extensions/v1beta1/NetworkPolicy (legacy) and diff --git a/npc/controller.go b/npc/controller.go index 304ca04a..6bd26b1e 100644 --- a/npc/controller.go +++ b/npc/controller.go @@ -9,7 +9,7 @@ import ( networkingv1 "k8s.io/api/networking/v1" "github.com/weaveworks/weave/common" - "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/net/ipset" "github.com/weaveworks/weave/npc/iptables" ) diff --git a/npc/controller_test.go b/npc/controller_test.go index 6174849d..f7dd7fdb 100644 --- a/npc/controller_test.go +++ b/npc/controller_test.go @@ -6,7 +6,7 @@ import ( "github.com/pkg/errors" "github.com/stretchr/testify/require" - "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/net/ipset" coreapi "k8s.io/api/core/v1" networkingv1 "k8s.io/api/networking/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" diff --git a/npc/namespace.go b/npc/namespace.go index 39cac4ca..5dc3f46e 100644 --- a/npc/namespace.go +++ b/npc/namespace.go @@ -13,7 +13,7 @@ import ( "k8s.io/apimachinery/pkg/util/uuid" "github.com/weaveworks/weave/common" - "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/net/ipset" "github.com/weaveworks/weave/npc/iptables" ) diff --git a/npc/selector.go b/npc/selector.go index 35a53f28..2a40e185 100644 --- a/npc/selector.go +++ b/npc/selector.go @@ -6,7 +6,7 @@ import ( "k8s.io/apimachinery/pkg/types" "github.com/weaveworks/weave/common" - "github.com/weaveworks/weave/npc/ipset" + "github.com/weaveworks/weave/net/ipset" ) type selectorSpec struct { diff --git a/prog/weave-kube/launch.sh b/prog/weave-kube/launch.sh index 9ee11402..d56d5f82 100755 --- a/prog/weave-kube/launch.sh +++ b/prog/weave-kube/launch.sh @@ -66,6 +66,11 @@ if [ "${EXPECT_NPC}" = "0" ]; then WEAVE_NPC_OPTS="" fi +NO_MASQ_LOCAL_OPT="" +if [ -n "${NO_MASQ_LOCAL}" ]; then + NO_MASQ_LOCAL_OPT="--no-masq-local" +fi + # Kubernetes sets HOSTNAME to the host's hostname # when running a pod in host namespace. NICKNAME_ARG="" @@ -156,5 +161,6 @@ post_start_actions & --ipalloc-init $IPALLOC_INIT \ --conn-limit=$CONN_LIMIT \ $WEAVE_NPC_OPTS \ + $NO_MASQ_LOCAL_OPT \ "$@" \ $KUBE_PEERS diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index 01b5266c..d5ffb17b 100644 --- a/prog/weave-npc/main.go +++ b/prog/weave-npc/main.go @@ -18,8 +18,8 @@ import ( "k8s.io/client-go/tools/cache" "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/net/ipset" "github.com/weaveworks/weave/npc" - "github.com/weaveworks/weave/npc/ipset" "github.com/weaveworks/weave/npc/metrics" "github.com/weaveworks/weave/npc/ulogd" ) @@ -56,7 +56,7 @@ func resetIPTables(ipt *iptables.IPTables) error { } func resetIPSets(ips ipset.Interface) error { - // Remove ipsets prefixed `weave-` only + // Remove ipsets prefixed `weave-` only. sets, err := ips.List(npc.IpsetNamePrefix) if err != nil { diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 52a7b6bc..f91ff893 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -26,6 +26,7 @@ import ( "github.com/weaveworks/weave/nameserver" weavenet "github.com/weaveworks/weave/net" "github.com/weaveworks/weave/net/address" + "github.com/weaveworks/weave/net/ipset" "github.com/weaveworks/weave/plugin" weaveproxy "github.com/weaveworks/weave/proxy" weave "github.com/weaveworks/weave/router" @@ -221,6 +222,7 @@ func main() { mflag.BoolVar(&pluginConfig.EnableV2Multicast, []string{"-plugin-v2-multicast"}, false, "enable multicast for Docker plugin (v2)") mflag.StringVar(&pluginConfig.Socket, []string{"-plugin-socket"}, "/run/docker/plugins/weave.sock", "plugin socket on which to listen") mflag.StringVar(&pluginConfig.MeshSocket, []string{"-plugin-mesh-socket"}, "/run/docker/plugins/weavemesh.sock", "plugin socket on which to listen in mesh mode") + mflag.BoolVar(&bridgeConfig.NoMasqLocal, []string{"-no-masq-local"}, false, "do not SNAT external traffic sent to containers running on this node") proxyConfig := newProxyConfig() @@ -305,7 +307,8 @@ func main() { bridgeConfig.Mac = name.String() bridgeConfig.Port = config.Port - bridgeType, err := weavenet.EnsureBridge(procPath, &bridgeConfig, Log) + ips := ipset.New(common.LogLogger()) + bridgeType, err := weavenet.EnsureBridge(procPath, &bridgeConfig, Log, ips) checkFatal(err) Log.Println("Bridge type is", bridgeType) @@ -336,6 +339,9 @@ func main() { if bridgeConfig.AWSVPC && !ipamConfig.Enabled() { Log.Fatalf("--awsvpc mode requires IPAM enabled") } + if bridgeConfig.AWSVPC && bridgeConfig.NoMasqLocal { + Log.Fatalf("--awsvpc mode is not compatible with the --no-masq-local option") + } db, err := db.NewBoltDB(dbPrefix) checkFatal(err) @@ -395,17 +401,19 @@ func main() { var ( allocator *ipam.Allocator defaultSubnet address.CIDR - trackerName string ) if ipamConfig.Enabled() { var t tracker.LocalRangeTracker if bridgeConfig.AWSVPC { - Log.Infoln("Creating AWSVPC LocalRangeTracker") t, err = tracker.NewAWSVPCTracker(bridgeConfig.WeaveBridgeName) if err != nil { Log.Fatalf("Cannot create AWSVPC LocalRangeTracker: %s", err) } - trackerName = "awsvpc" + } else if bridgeConfig.NoMasqLocal { + t = weavenet.NewNoMasqLocalTracker(ips) + } + if t != nil { + Log.Infof("Using %q LocalRangeTracker", t) } preClaims, err := findExistingAddresses(dockerCli, bridgeConfig.WeaveBridgeName) @@ -453,7 +461,7 @@ func main() { if httpAddr != "" { muxRouter := mux.NewRouter() if allocator != nil { - allocator.HandleHTTP(muxRouter, defaultSubnet, trackerName, dockerCli) + allocator.HandleHTTP(muxRouter, defaultSubnet, dockerCli) } if ns != nil { ns.HandleHTTP(muxRouter, dockerCli) diff --git a/site/kubernetes/kube-addon.md b/site/kubernetes/kube-addon.md index fdc6bd61..383654de 100644 --- a/site/kubernetes/kube-addon.md +++ b/site/kubernetes/kube-addon.md @@ -364,6 +364,9 @@ The list of variables you can set is: a larger size for better performance if your network supports jumbo frames - see [here](/site/tasks/manage/fastdp.md#mtu) for more details. +* `NO_MASQ_LOCAL` - set to 1 to preserve the client source IP address when + accessing Service annotated with `service.spec.externalTrafficPolicy=Local`. + The feature works only with Weave IPAM (default). Example: ``` diff --git a/test/840_weave_kube_3_test.sh b/test/840_weave_kube_3_test.sh index 63d305bf..4feab89d 100755 --- a/test/840_weave_kube_3_test.sh +++ b/test/840_weave_kube_3_test.sh @@ -44,14 +44,14 @@ for host in $HOSTS; do done if [ -n "$COVERAGE" ]; then - COVERAGE_ARGS="env:\\n - name: EXTRA_ARGS\\n value: \"-test.coverprofile=/home/weave/cover.prof --\"" + WEAVE_ENV_VARS="env:\\n - name: EXTRA_ARGS\\n value: \"-test.coverprofile=/home/weave/cover.prof --\"" else - COVERAGE_ARGS="env:" + WEAVE_ENV_VARS="env:" fi # Ensure Kubernetes uses locally built container images and inject code coverage environment variable (or do nothing depending on $COVERAGE): sed -e "s%imagePullPolicy: Always%imagePullPolicy: Never%" \ - -e "s%env:%$COVERAGE_ARGS%" \ + -e "s%env:%$WEAVE_ENV_VARS%" \ -e "s%#npc-args% args:\n - '--use-legacy-netpol'%" \ "$(dirname "$0")/../prog/weave-kube/weave-daemonset-k8s-1.7.yaml" | run_on "$HOST1" "$KUBECTL apply -n kube-system -f -" @@ -63,7 +63,7 @@ check_connections() { assert_raises 'wait_for_x check_connections "connections to establish"' -# Check we can ping between the Weave bridg IPs on each host +# Check we can ping between the Weave bridge IPs on each host HOST1EXPIP=$($SSH $HOST1 "weave expose" || true) HOST2EXPIP=$($SSH $HOST2 "weave expose" || true) HOST3EXPIP=$($SSH $HOST3 "weave expose" || true) @@ -182,7 +182,7 @@ assert_raises "! $SSH $HOST1 $KUBECTL exec $denyPodName -- curl -s -S -f -m 2 ht # Restart weave-net with npc in non-legacy mode $SSH $HOST1 "$KUBECTL delete ds weave-net -n=kube-system" sed -e "s%imagePullPolicy: Always%imagePullPolicy: Never%" \ - -e "s%env:%$COVERAGE_ARGS%" \ + -e "s%env:%$WEAVE_ENV_VARS%" \ "$(dirname "$0")/../prog/weave-kube/weave-daemonset-k8s-1.7.yaml" | run_on "$HOST1" "$KUBECTL apply -n kube-system -f -" assert_raises 'wait_for_x check_all_pods_communicate pods' @@ -244,6 +244,26 @@ assert_raises "$SSH $HOST1 $KUBECTL exec $denyPodName -- curl -s -S -f -m 2 http assert_raises "$SSH $HOST1 curl -s -S -f -m 2 http://$VIRTUAL_IP/status >/dev/null" assert_raises "$SSH $HOST1 curl -s -S -f -m 2 http://$HOST2:31138/status >/dev/null" +# Passing --no-masq-local and setting externalTrafficPolicy to Local must preserve +# the client source IP addr + +CLIENT_IP_MASQ="$($SSH $HOST1 curl -sS http://$HOST2:31138/client_ip)" + +WEAVE_ENV_VARS="${WEAVE_ENV_VARS}\\n - name: NO_MASQ_LOCAL\\n value: \"1\"" +$SSH $HOST1 "$KUBECTL delete ds weave-net -n=kube-system" +sed -e "s%imagePullPolicy: Always%imagePullPolicy: Never%" \ + -e "s%env:%$WEAVE_ENV_VARS%" \ + "$(dirname "$0")/../prog/weave-kube/weave-daemonset-k8s-1.7.yaml" | run_on "$HOST1" "$KUBECTL apply -n kube-system -f -" + +sleep 5 + +run_on $HOST1 "$KUBECTL patch svc netvirt -p '{\"spec\":{\"externalTrafficPolicy\":\"Local\"}}'" + +CLIENT_IP_NO_MASQ="$($SSH $HOST1 curl -sS http://$HOST2:31138/client_ip)" + +assert_raises "[ $CLIENT_IP_NO_MASQ != $CLIENT_IP_MASQ ]" +assert_raises "[ $HOST1IP == $CLIENT_IP_NO_MASQ ]" + tear_down_kubeadm # Destroy our test ipset, and implicitly check it is still there diff --git a/test/images/network-tester/webserver.go b/test/images/network-tester/webserver.go index e27fd0e6..4315a2aa 100644 --- a/test/images/network-tester/webserver.go +++ b/test/images/network-tester/webserver.go @@ -13,6 +13,8 @@ // Visit /status to see pass/running/fail determination. (literally, it will // return one of those words.) // +// Visit /client_ip to see the source IP addr of the request. +// // /write is used by other network test servers to register connectivity. package main @@ -78,6 +80,20 @@ func (s *State) serveStatus(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "fail") } +// serveClientIP returns the client source IP addr. +func (s *State) serveClientIP(w http.ResponseWriter, r *http.Request) { + s.lock.Lock() + defer s.lock.Unlock() + + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + go s.Logf("Warning: unable to parse remote addr: %s", err) + w.WriteHeader(http.StatusInternalServerError) + } + + fmt.Fprintf(w, host) +} + // serveRead writes our json encoded state func (s *State) serveRead(w http.ResponseWriter, r *http.Request) { s.lock.Lock() @@ -186,6 +202,7 @@ func main() { http.HandleFunc("/read", state.serveRead) http.HandleFunc("/write", state.serveWrite) http.HandleFunc("/status", state.serveStatus) + http.HandleFunc("/client_ip", state.serveClientIP) go log.Fatal(http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", *port), nil)) diff --git a/test/run_all.sh b/test/run_all.sh index 51c674a7..1d6b500f 100755 --- a/test/run_all.sh +++ b/test/run_all.sh @@ -23,4 +23,4 @@ if [ -n "$CIRCLECI" -o -n "$PARALLEL" ]; then RUNNER_ARGS="$RUNNER_ARGS -parallel" fi -HOSTS="$HOSTS" "${DIR}/../tools/runner/runner" -timeout=300 $RUNNER_ARGS $TESTS +HOSTS="$HOSTS" "${DIR}/../tools/runner/runner" -timeout=360 $RUNNER_ARGS $TESTS diff --git a/weave b/weave index f644e4ea..8490832b 100755 --- a/weave +++ b/weave @@ -505,6 +505,8 @@ destroy_bridge() { run_iptables -t nat -D POSTROUTING -j WEAVE >/dev/null 2>&1 || true run_iptables -t nat -D POSTROUTING -o $BRIDGE -j ACCEPT >/dev/null 2>&1 || true run_iptables -t nat -X WEAVE >/dev/null 2>&1 || true + + ipset destroy weaver-no-masq-local >/dev/null 2>&1 || true } add_iface_fastdp() {