Move test gossip router out of ipam code for use in gossip dns code.
This commit is contained in:
1 parent
e6828e1aa5
commit
89ad7e1ba2
3 files changed
+141
-126
No files matched your search
+11
-10
@@ -8,9 +8,10 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/weaveworks/weave/common"
|
||||
"github.com/weaveworks/weave/net/address"
|
||||
"github.com/weaveworks/weave/router"
|
||||
"github.com/weaveworks/weave/testing/gossip"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -166,13 +167,13 @@ func TestCancel(t *testing.T) {
|
||||
CIDR = "10.0.1.7/26"
|
||||
)
|
||||
|
||||
router := TestGossipRouter{make(map[router.PeerName]chan interface{}), 0.0}
|
||||
router := gossip.NewTestRouter(0.0)
|
||||
|
||||
alloc1, subnet := makeAllocator("01:00:00:02:00:00", CIDR, 2)
|
||||
alloc1.SetInterfaces(router.connect(alloc1.ourName, alloc1))
|
||||
alloc1.SetInterfaces(router.Connect(alloc1.ourName, alloc1))
|
||||
|
||||
alloc2, _ := makeAllocator("02:00:00:02:00:00", CIDR, 2)
|
||||
alloc2.SetInterfaces(router.connect(alloc2.ourName, alloc2))
|
||||
alloc2.SetInterfaces(router.Connect(alloc2.ourName, alloc2))
|
||||
alloc1.claimRingForTesting(alloc1, alloc2)
|
||||
alloc2.claimRingForTesting(alloc1, alloc2)
|
||||
|
||||
@@ -254,17 +255,17 @@ func TestTransfer(t *testing.T) {
|
||||
require.True(t, err == nil, "Failed to get address")
|
||||
|
||||
router.GossipBroadcast(alloc2.Gossip())
|
||||
router.flush()
|
||||
router.Flush()
|
||||
router.GossipBroadcast(alloc3.Gossip())
|
||||
router.flush()
|
||||
router.removePeer(alloc2.ourName)
|
||||
router.removePeer(alloc3.ourName)
|
||||
router.Flush()
|
||||
router.RemovePeer(alloc2.ourName)
|
||||
router.RemovePeer(alloc3.ourName)
|
||||
alloc2.Stop()
|
||||
alloc3.Stop()
|
||||
router.flush()
|
||||
router.Flush()
|
||||
require.NoError(t, alloc1.AdminTakeoverRanges(alloc2.ourName.String()))
|
||||
require.NoError(t, alloc1.AdminTakeoverRanges(alloc3.ourName.String()))
|
||||
router.flush()
|
||||
router.Flush()
|
||||
|
||||
require.Equal(t, address.Offset(1022), alloc1.NumFreeAddresses(subnet))
|
||||
|
||||
|
||||
+5
-116
@@ -8,9 +8,9 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/weaveworks/weave/common"
|
||||
"github.com/weaveworks/weave/net/address"
|
||||
"github.com/weaveworks/weave/router"
|
||||
"github.com/weaveworks/weave/testing/gossip"
|
||||
)
|
||||
|
||||
type mockMessage struct {
|
||||
@@ -194,119 +194,8 @@ func AssertNothingSentErr(t *testing.T, ch <-chan error) {
|
||||
}
|
||||
}
|
||||
|
||||
// Router to convey gossip from one gossiper to another, for testing
|
||||
type unicastMessage struct {
|
||||
sender router.PeerName
|
||||
buf []byte
|
||||
}
|
||||
type broadcastMessage struct {
|
||||
data router.GossipData
|
||||
}
|
||||
type exitMessage struct {
|
||||
exitChan chan struct{}
|
||||
}
|
||||
type flushMessage struct {
|
||||
flushChan chan struct{}
|
||||
}
|
||||
|
||||
type TestGossipRouter struct {
|
||||
gossipChans map[router.PeerName]chan interface{}
|
||||
loss float32 // 0.0 means no loss
|
||||
}
|
||||
|
||||
func (grouter *TestGossipRouter) GossipBroadcast(update router.GossipData) error {
|
||||
for _, gossipChan := range grouter.gossipChans {
|
||||
select {
|
||||
case gossipChan <- broadcastMessage{data: update}:
|
||||
default: // drop the message if we cannot send it
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (grouter *TestGossipRouter) flush() {
|
||||
for _, gossipChan := range grouter.gossipChans {
|
||||
flushChan := make(chan struct{})
|
||||
gossipChan <- flushMessage{flushChan: flushChan}
|
||||
<-flushChan
|
||||
}
|
||||
}
|
||||
|
||||
func (grouter *TestGossipRouter) removePeer(peer router.PeerName) {
|
||||
gossipChan := grouter.gossipChans[peer]
|
||||
resultChan := make(chan struct{})
|
||||
gossipChan <- exitMessage{exitChan: resultChan}
|
||||
<-resultChan
|
||||
delete(grouter.gossipChans, peer)
|
||||
}
|
||||
|
||||
type TestGossipRouterClient struct {
|
||||
router *TestGossipRouter
|
||||
sender router.PeerName
|
||||
}
|
||||
|
||||
func (grouter *TestGossipRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) {
|
||||
gossipTimer := time.Tick(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case gossip := <-gossipChan:
|
||||
switch message := gossip.(type) {
|
||||
case exitMessage:
|
||||
close(message.exitChan)
|
||||
return
|
||||
|
||||
case flushMessage:
|
||||
close(message.flushChan)
|
||||
|
||||
case unicastMessage:
|
||||
if rand.Float32() > (1.0 - grouter.loss) {
|
||||
continue
|
||||
}
|
||||
if err := gossiper.OnGossipUnicast(message.sender, message.buf); err != nil {
|
||||
panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", message.sender, err))
|
||||
}
|
||||
|
||||
case broadcastMessage:
|
||||
if rand.Float32() > (1.0 - grouter.loss) {
|
||||
continue
|
||||
}
|
||||
for _, msg := range message.data.Encode() {
|
||||
if _, err := gossiper.OnGossipBroadcast(msg); err != nil {
|
||||
panic(fmt.Sprintf("Error doing gossip broadcast: %s", err))
|
||||
}
|
||||
}
|
||||
}
|
||||
case <-gossipTimer:
|
||||
grouter.GossipBroadcast(gossiper.Gossip())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip {
|
||||
gossipChan := make(chan interface{}, 100)
|
||||
|
||||
go grouter.run(gossiper, gossipChan)
|
||||
|
||||
grouter.gossipChans[sender] = gossipChan
|
||||
return TestGossipRouterClient{grouter, sender}
|
||||
}
|
||||
|
||||
func (client TestGossipRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []byte) error {
|
||||
select {
|
||||
case client.router.gossipChans[dstPeerName] <- unicastMessage{sender: client.sender, buf: buf}:
|
||||
default: // drop the message if we cannot send it
|
||||
common.Log.Errorf("Dropping message")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (client TestGossipRouterClient) GossipBroadcast(update router.GossipData) error {
|
||||
return client.router.GossipBroadcast(update)
|
||||
}
|
||||
|
||||
func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, TestGossipRouter, address.Range) {
|
||||
|
||||
gossipRouter := TestGossipRouter{make(map[router.PeerName]chan interface{}), 0.0}
|
||||
func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, *gossip.TestRouter, address.Range) {
|
||||
gossipRouter := gossip.NewTestRouter(0.0)
|
||||
allocs := make([]*Allocator, size)
|
||||
var subnet address.Range
|
||||
|
||||
@@ -314,13 +203,13 @@ func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, TestGossipRou
|
||||
var alloc *Allocator
|
||||
alloc, subnet = makeAllocator(fmt.Sprintf("%02d:00:00:02:00:00", i),
|
||||
cidr, uint(size/2+1))
|
||||
alloc.SetInterfaces(gossipRouter.connect(alloc.ourName, alloc))
|
||||
alloc.SetInterfaces(gossipRouter.Connect(alloc.ourName, alloc))
|
||||
alloc.Start()
|
||||
allocs[i] = alloc
|
||||
}
|
||||
|
||||
gossipRouter.GossipBroadcast(allocs[size-1].Gossip())
|
||||
gossipRouter.flush()
|
||||
gossipRouter.Flush()
|
||||
return allocs, gossipRouter, subnet
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package gossip
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"time"
|
||||
|
||||
"github.com/weaveworks/weave/common"
|
||||
"github.com/weaveworks/weave/router"
|
||||
)
|
||||
|
||||
// Router to convey gossip from one gossiper to another, for testing
|
||||
type unicastMessage struct {
|
||||
sender router.PeerName
|
||||
buf []byte
|
||||
}
|
||||
type broadcastMessage struct {
|
||||
data router.GossipData
|
||||
}
|
||||
type exitMessage struct {
|
||||
exitChan chan struct{}
|
||||
}
|
||||
type flushMessage struct {
|
||||
flushChan chan struct{}
|
||||
}
|
||||
|
||||
type TestRouter struct {
|
||||
gossipChans map[router.PeerName]chan interface{}
|
||||
loss float32 // 0.0 means no loss
|
||||
}
|
||||
|
||||
func NewTestRouter(loss float32) *TestRouter {
|
||||
return &TestRouter{make(map[router.PeerName]chan interface{}, 100), loss}
|
||||
}
|
||||
|
||||
func (grouter *TestRouter) GossipBroadcast(update router.GossipData) error {
|
||||
for _, gossipChan := range grouter.gossipChans {
|
||||
select {
|
||||
case gossipChan <- broadcastMessage{data: update}:
|
||||
default: // drop the message if we cannot send it
|
||||
common.Log.Errorf("Dropping message")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (grouter *TestRouter) Flush() {
|
||||
for _, gossipChan := range grouter.gossipChans {
|
||||
flushChan := make(chan struct{})
|
||||
gossipChan <- flushMessage{flushChan: flushChan}
|
||||
<-flushChan
|
||||
}
|
||||
}
|
||||
|
||||
func (grouter *TestRouter) RemovePeer(peer router.PeerName) {
|
||||
gossipChan := grouter.gossipChans[peer]
|
||||
resultChan := make(chan struct{})
|
||||
gossipChan <- exitMessage{exitChan: resultChan}
|
||||
<-resultChan
|
||||
delete(grouter.gossipChans, peer)
|
||||
}
|
||||
|
||||
type TestRouterClient struct {
|
||||
router *TestRouter
|
||||
sender router.PeerName
|
||||
}
|
||||
|
||||
func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) {
|
||||
gossipTimer := time.Tick(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case gossip := <-gossipChan:
|
||||
switch message := gossip.(type) {
|
||||
case exitMessage:
|
||||
close(message.exitChan)
|
||||
return
|
||||
|
||||
case flushMessage:
|
||||
close(message.flushChan)
|
||||
|
||||
case unicastMessage:
|
||||
if rand.Float32() > (1.0 - grouter.loss) {
|
||||
continue
|
||||
}
|
||||
if err := gossiper.OnGossipUnicast(message.sender, message.buf); err != nil {
|
||||
panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", message.sender, err))
|
||||
}
|
||||
|
||||
case broadcastMessage:
|
||||
if rand.Float32() > (1.0 - grouter.loss) {
|
||||
continue
|
||||
}
|
||||
for _, msg := range message.data.Encode() {
|
||||
if _, err := gossiper.OnGossipBroadcast(msg); err != nil {
|
||||
panic(fmt.Sprintf("Error doing gossip broadcast: %s", err))
|
||||
}
|
||||
}
|
||||
}
|
||||
case <-gossipTimer:
|
||||
grouter.GossipBroadcast(gossiper.Gossip())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (grouter *TestRouter) Connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip {
|
||||
gossipChan := make(chan interface{}, 100)
|
||||
|
||||
go grouter.run(gossiper, gossipChan)
|
||||
|
||||
grouter.gossipChans[sender] = gossipChan
|
||||
return TestRouterClient{grouter, sender}
|
||||
}
|
||||
|
||||
func (client TestRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []byte) error {
|
||||
select {
|
||||
case client.router.gossipChans[dstPeerName] <- unicastMessage{sender: client.sender, buf: buf}:
|
||||
default: // drop the message if we cannot send it
|
||||
common.Log.Errorf("Dropping message")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (client TestRouterClient) GossipBroadcast(update router.GossipData) error {
|
||||
return client.router.GossipBroadcast(update)
|
||||
}
|
||||
Reference in new issue
Block a user