Merge pull request #3298 from weaveworks/issues/2924-preserve-src-ip-on-local

2924: Preserve the client source IP

.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

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

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))

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())
 	})
 }

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 {

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 {

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
+}

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

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
 }

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
+}

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

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

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"
 )
 

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"

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"
 )
 

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 {

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

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 {

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)

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:
 ```

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

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))
 

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

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() {