Ignore own lease in subnet events

Add an option to ignore a lease in WatchLeases.
Backends can pass their own lease so filter out events
related to them.

Fixes #241 and #126

backend/hostgw/hostgw.go

@@ -101,7 +101,7 @@ func (rb *HostgwBackend) Run() {
 	evts := make(chan []subnet.Event)
 	rb.wg.Add(1)
 	go func() {
-		subnet.WatchLeases(rb.ctx, rb.sm, rb.network, evts)
+		subnet.WatchLeases(rb.ctx, rb.sm, rb.network, rb.lease, evts)
 		rb.wg.Done()
 	}()
 
@@ -149,9 +149,6 @@ func (rb *HostgwBackend) handleSubnetEvents(batch []subnet.Event) {
 				Gw:        evt.Lease.Attrs.PublicIP.ToIP(),
 				LinkIndex: rb.extIface.Index,
 			}
-			if rb.extIaddr.Equal(route.Gw) {
-				continue
-			}
 			if err := netlink.RouteAdd(&route); err != nil {
 				log.Errorf("Error adding route to %v via %v: %v", evt.Lease.Subnet, evt.Lease.Attrs.PublicIP, err)
 				continue

backend/udp/udp.go

@@ -225,7 +225,7 @@ func (m *UdpBackend) monitorEvents() {
 
 	m.wg.Add(1)
 	go func() {
-		subnet.WatchLeases(m.ctx, m.sm, m.network, evts)
+		subnet.WatchLeases(m.ctx, m.sm, m.network, m.lease, evts)
 		m.wg.Done()
 	}()
 

backend/vxlan/vxlan.go

@@ -160,7 +160,7 @@ func (vb *VXLANBackend) Run() {
 	evts := make(chan []subnet.Event)
 	vb.wg.Add(1)
 	go func() {
-		subnet.WatchLeases(vb.ctx, vb.sm, vb.network, evts)
+		subnet.WatchLeases(vb.ctx, vb.sm, vb.network, vb.lease, evts)
 		log.Info("WatchLeases exited")
 		vb.wg.Done()
 	}()

remote/remote_test.go

@@ -134,7 +134,7 @@ func doTestWatch(t *testing.T, sm subnet.Manager) {
 
 	events := make(chan []subnet.Event)
 	go func() {
-		subnet.WatchLeases(ctx, sm, "_", events)
+		subnet.WatchLeases(ctx, sm, "_", nil, events)
 		wg.Done()
 	}()
 

subnet/subnet_test.go

@@ -111,6 +111,20 @@ func newIP4Net(ipaddr string, prefix uint) ip.IP4Net {
 	}
 }
 
+func acquireLease(ctx context.Context, t *testing.T, sm Manager) *Lease {
+	extIaddr, _ := ip.ParseIP4("1.2.3.4")
+	attrs := LeaseAttrs{
+		PublicIP: extIaddr,
+	}
+
+	l, err := sm.AcquireLease(ctx, "", &attrs)
+	if err != nil {
+		t.Fatal("AcquireLease failed: ", err)
+	}
+
+	return l
+}
+
 func TestWatchLeaseAdded(t *testing.T) {
 	msr := newDummyRegistry(0)
 	sm := newEtcdManager(msr)
@@ -118,25 +132,31 @@ func TestWatchLeaseAdded(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	l := acquireLease(ctx, t, sm)
+
 	events := make(chan []Event)
-	go WatchLeases(ctx, sm, "", events)
+	go WatchLeases(ctx, sm, "", l, events)
 
-	// skip over the initial snapshot
-	<-events
+	evtBatch := <-events
+	for _, evt := range evtBatch {
+		if evt.Lease.Key() == l.Key() {
+			t.Errorf("WatchLeases returned our own lease")
+		}
+	}
 
-	expected := "10.3.3.0-24"
+	expected := "10.3.6.0-24"
 	msr.createSubnet(ctx, "_", expected, `{"PublicIP": "1.1.1.1"}`, 0)
 
-	evtBatch := <-events
+	evtBatch = <-events
 
 	if len(evtBatch) != 1 {
-		t.Fatalf("WatchSubnets produced wrong sized event batch")
+		t.Fatalf("WatchLeases produced wrong sized event batch")
 	}
 
 	evt := evtBatch[0]
 
 	if evt.Type != SubnetAdded {
-		t.Fatalf("WatchSubnets produced wrong event type")
+		t.Fatalf("WatchLeases produced wrong event type")
 	}
 
 	actual := evt.Lease.Key()
@@ -152,25 +172,30 @@ func TestWatchLeaseRemoved(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	l := acquireLease(ctx, t, sm)
+
 	events := make(chan []Event)
-	go WatchLeases(ctx, sm, "", events)
+	go WatchLeases(ctx, sm, "", l, events)
 
-	// skip over the initial snapshot
-	<-events
+	evtBatch := <-events
+	for _, evt := range evtBatch {
+		if evt.Lease.Key() == l.Key() {
+			t.Errorf("WatchLeases returned our own lease")
+		}
+	}
 
 	expected := "10.3.4.0-24"
 	msr.expireSubnet(expected)
 
-	evtBatch := <-events
-
+	evtBatch = <-events
 	if len(evtBatch) != 1 {
-		t.Fatalf("WatchSubnets produced wrong sized event batch")
+		t.Fatalf("WatchLeases produced wrong sized event batch")
 	}
 
 	evt := evtBatch[0]
 
 	if evt.Type != SubnetRemoved {
-		t.Fatalf("WatchSubnets produced wrong event type")
+		t.Fatalf("WatchLeases produced wrong event type")
 	}
 
 	actual := evt.Lease.Key()

subnet/watch.go

@@ -25,8 +25,10 @@ import (
 // and communicates addition/deletion events on receiver channel. It takes care
 // of handling "fall-behind" logic where the history window has advanced too far
 // and it needs to diff the latest snapshot with its saved state and generate events
-func WatchLeases(ctx context.Context, sm Manager, network string, receiver chan []Event) {
-	lw := &leaseWatcher{}
+func WatchLeases(ctx context.Context, sm Manager, network string, ownLease *Lease, receiver chan []Event) {
+	lw := &leaseWatcher{
+		ownLease: ownLease,
+	}
 	var cursor interface{}
 
 	for {
@@ -57,13 +59,18 @@ func WatchLeases(ctx context.Context, sm Manager, network string, receiver chan
 }
 
 type leaseWatcher struct {
-	leases []Lease
+	ownLease *Lease
+	leases   []Lease
 }
 
 func (lw *leaseWatcher) reset(leases []Lease) []Event {
 	batch := []Event{}
 
 	for _, nl := range leases {
+		if lw.ownLease != nil && nl.Subnet.Equal(lw.ownLease.Subnet) {
+			continue
+		}
+
 		found := false
 		for i, ol := range lw.leases {
 			if ol.Subnet.Equal(nl.Subnet) {
@@ -93,6 +100,10 @@ func (lw *leaseWatcher) update(events []Event) []Event {
 	batch := []Event{}
 
 	for _, e := range events {
+		if lw.ownLease != nil && e.Lease.Subnet.Equal(lw.ownLease.Subnet) {
+			continue
+		}
+
 		switch e.Type {
 		case SubnetAdded:
 			batch = append(batch, lw.add(&e.Lease))