reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '1.2'

Browse Source
This commit is contained in:
Matthias Radestock committed 2015-11-02 20:58:15 +00:00
2 parents 6a238818da 77c7963bde
commit 30a128c00b
5 files changed
+109 -82
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

nameserver/entry.go
+34 -10
View File
@@ -18,7 +18,8 @@ type Entry struct {
ContainerID string
Origin router.PeerName
Addr address.Address
Hostname string
Hostname string // as supplied
lHostname string // lowercased (not exported, so not encoded by gob)
Version int
Tombstone int64 // timestamp of when it was deleted
}
@@ -69,7 +70,7 @@ func (e1 *Entry) less(e2 *Entry) bool {
func (e1 *Entry) insensitiveLess(e2 *Entry) bool {
// Entries are kept sorted by Hostname, Origin, ContainerID then address
e1Hostname, e2Hostname := strings.ToLower(e1.Hostname), strings.ToLower(e2.Hostname)
e1Hostname, e2Hostname := e1.lHostname, e2.lHostname
switch {
case e1Hostname != e2Hostname:
return e1Hostname < e2Hostname
@@ -100,6 +101,10 @@ func (e1 *Entry) String() string {
return fmt.Sprintf("%s -> %s", e1.Hostname, e1.Addr.String())
}
func (e1 *Entry) addLowercase() {
e1.lHostname = strings.ToLower(e1.Hostname)
}
func check(es SortableEntries) error {
if !sort.IsSorted(es) {
return fmt.Errorf("Not sorted!")
@@ -126,7 +131,8 @@ func (es *Entries) checkAndPanic() *Entries {
func (es *Entries) add(hostname, containerid string, origin router.PeerName, addr address.Address) Entry {
defer es.checkAndPanic().checkAndPanic()
entry := Entry{Hostname: hostname, Origin: origin, ContainerID: containerid, Addr: addr}
entry := Entry{Hostname: hostname, lHostname: strings.ToLower(hostname),
Origin: origin, ContainerID: containerid, Addr: addr}
i := sort.Search(len(*es), func(i int) bool {
return !(*es)[i].insensitiveLess(&entry)
})
@@ -145,6 +151,7 @@ func (es *Entries) add(hostname, containerid string, origin router.PeerName, add
func (es *Entries) merge(incoming Entries) Entries {
defer es.checkAndPanic().checkAndPanic()
incoming.checkAndPanic()
newEntries := Entries{}
i := 0
@@ -203,14 +210,14 @@ func (es Entries) lookup(hostname string) Entries {
lowerHostname := strings.ToLower(hostname)
i := sort.Search(len(es), func(i int) bool {
return strings.ToLower(es[i].Hostname) >= lowerHostname
return es[i].lHostname >= lowerHostname
})
if i >= len(es) || strings.ToLower(es[i].Hostname) != lowerHostname {
if i >= len(es) || es[i].lHostname != lowerHostname {
return Entries{}
}
j := sort.Search(len(es)-i, func(j int) bool {
return strings.ToLower(es[i+j].Hostname) > lowerHostname
return es[i+j].lHostname > lowerHostname
})
return es[i : i+j]
@@ -227,14 +234,18 @@ func (es Entries) first(f func(*Entry) bool) (*Entry, error) {
return nil, fmt.Errorf("Not found")
}
func (es Entries) addLowercase() {
for i := range es {
es[i].addLowercase()
}
}
type GossipData struct {
Timestamp int64
Entries
}
func (g *GossipData) Merge(o router.GossipData) {
checkAndPanic(CaseSensitive(g.Entries))
defer func() { checkAndPanic(CaseSensitive(g.Entries)) }()
other := o.(*GossipData)
g.Entries.merge(other.Entries)
if g.Timestamp < other.Timestamp {
@@ -242,10 +253,23 @@ func (g *GossipData) Merge(o router.GossipData) {
}
}
func (g *GossipData) Decode(msg []byte) error {
if err := gob.NewDecoder(bytes.NewReader(msg)).Decode(g); err != nil {
return err
}
g.Entries.addLowercase() // lowercase strings not sent on the wire
sort.Sort(CaseInsensitive(g.Entries))
return nil
}
func (g *GossipData) Encode() [][]byte {
checkAndPanic(CaseSensitive(g.Entries))
// Make a copy so we can sort: all outgoing data is sent in case-sensitive order
g2 := GossipData{Timestamp: g.Timestamp, Entries: make(Entries, len(g.Entries))}
copy(g2.Entries, g.Entries)
sort.Sort(CaseSensitive(g2.Entries))
buf := &bytes.Buffer{}
if err := gob.NewEncoder(buf).Encode(g); err != nil {
if err := gob.NewEncoder(buf).Encode(g2); err != nil {
panic(err)
}
return [][]byte{buf.Bytes()}
nameserver/entry_test.go
+47 -56
View File
@@ -9,6 +9,19 @@ import (
"github.com/weaveworks/weave/router"
)
func l(es Entries) Entries {
es.addLowercase()
return es
}
func makeEntries(values string) Entries {
entries := make(Entries, len(values))
for i, c := range values {
entries[i] = Entry{Hostname: string(c)}
}
return l(entries)
}
func TestAdd(t *testing.T) {
oldNow := now
defer func() { now = oldNow }()
@@ -16,68 +29,46 @@ func TestAdd(t *testing.T) {
entries := Entries{}
entries.add("A", "", router.UnknownPeerName, address.Address(0))
expected := Entries{
expected := l(Entries{
Entry{Hostname: "A", Origin: router.UnknownPeerName, Addr: address.Address(0)},
}
})
require.Equal(t, entries, expected)
entries.tombstone(router.UnknownPeerName, func(e *Entry) bool { return e.Hostname == "A" })
expected = Entries{
expected = l(Entries{
Entry{Hostname: "A", Origin: router.UnknownPeerName, Addr: address.Address(0), Version: 1, Tombstone: 1234},
}
})
require.Equal(t, entries, expected)
entries.add("A", "", router.UnknownPeerName, address.Address(0))
expected = Entries{
expected = l(Entries{
Entry{Hostname: "A", Origin: router.UnknownPeerName, Addr: address.Address(0), Version: 2},
}
})
require.Equal(t, entries, expected)
}
func TestMerge(t *testing.T) {
e1 := Entries{
Entry{Hostname: "A"},
Entry{Hostname: "C"},
Entry{Hostname: "D"},
Entry{Hostname: "F"},
}
e2 := Entries{
Entry{Hostname: "B"},
Entry{Hostname: "E"},
Entry{Hostname: "F"},
}
e1 := makeEntries("ACDF")
e2 := makeEntries("BEF")
diff := e1.merge(e2)
expectedDiff := Entries{
Entry{Hostname: "B"},
Entry{Hostname: "E"},
}
require.Equal(t, expectedDiff, diff)
expected := Entries{
Entry{Hostname: "A"},
Entry{Hostname: "B"},
Entry{Hostname: "C"},
Entry{Hostname: "D"},
Entry{Hostname: "E"},
Entry{Hostname: "F"},
}
require.Equal(t, expected, e1)
require.Equal(t, makeEntries("BE"), diff)
require.Equal(t, makeEntries("ABCDEF"), e1)
diff = e1.merge(e1)
require.Equal(t, Entries{}, diff)
}
func TestOldMerge(t *testing.T) {
e1 := Entries{Entry{Hostname: "A", Version: 0}}
diff := e1.merge(Entries{Entry{Hostname: "A", Version: 1}})
require.Equal(t, Entries{Entry{Hostname: "A", Version: 1}}, diff)
require.Equal(t, Entries{Entry{Hostname: "A", Version: 1}}, e1)
e1 := l(Entries{Entry{Hostname: "A", Version: 0}})
diff := e1.merge(l(Entries{Entry{Hostname: "A", Version: 1}}))
require.Equal(t, l(Entries{Entry{Hostname: "A", Version: 1}}), diff)
require.Equal(t, l(Entries{Entry{Hostname: "A", Version: 1}}), e1)
diff = e1.merge(Entries{Entry{Hostname: "A", Version: 0}})
diff = e1.merge(l(Entries{Entry{Hostname: "A", Version: 0}}))
require.Equal(t, Entries{}, diff)
require.Equal(t, Entries{Entry{Hostname: "A", Version: 1}}, e1)
require.Equal(t, l(Entries{Entry{Hostname: "A", Version: 1}}), e1)
}
func TestTombstone(t *testing.T) {
@@ -85,48 +76,48 @@ func TestTombstone(t *testing.T) {
defer func() { now = oldNow }()
now = func() int64 { return 1234 }
es := Entries{
Entry{Hostname: "A"},
Entry{Hostname: "B"},
}
es := makeEntries("AB")
es.tombstone(router.UnknownPeerName, func(e *Entry) bool {
return e.Hostname == "B"
})
expected := Entries{
expected := l(Entries{
Entry{Hostname: "A"},
Entry{Hostname: "B", Version: 1, Tombstone: 1234},
}
})
require.Equal(t, expected, es)
}
func TestDelete(t *testing.T) {
es := Entries{
Entry{Hostname: "A"},
Entry{Hostname: "B"},
}
es := makeEntries("AB")
es.filter(func(e *Entry) bool {
return e.Hostname != "A"
})
expected := Entries{
Entry{Hostname: "B"},
}
require.Equal(t, expected, es)
require.Equal(t, makeEntries("B"), es)
}
func TestLookup(t *testing.T) {
es := Entries{
es := l(Entries{
Entry{Hostname: "A"},
Entry{Hostname: "B", ContainerID: "bar"},
Entry{Hostname: "B", ContainerID: "foo"},
Entry{Hostname: "C"},
}
})
have := es.lookup("B")
want := Entries{
want := l(Entries{
Entry{Hostname: "B", ContainerID: "bar"},
Entry{Hostname: "B", ContainerID: "foo"},
}
})
require.Equal(t, have, want)
}
func TestGossipDataMerge(t *testing.T) {
g1 := GossipData{Entries: makeEntries("AcDf")}
g2 := GossipData{Entries: makeEntries("BEf")}
g1.Merge(&g2)
require.Equal(t, GossipData{Entries: makeEntries("ABcDEf")}, g1)
}
nameserver/nameserver.go
+1 -8
View File
@@ -1,10 +1,7 @@
package nameserver
import (
"bytes"
"encoding/gob"
"fmt"
"sort"
"sync"
"time"
@@ -192,7 +189,6 @@ func (n *Nameserver) Gossip() router.GossipData {
Timestamp: now(),
}
copy(gossip.Entries, n.entries)
sort.Sort(CaseSensitive(gossip.Entries))
return gossip
}
@@ -202,16 +198,13 @@ func (n *Nameserver) OnGossipUnicast(sender router.PeerName, msg []byte) error {
func (n *Nameserver) receiveGossip(msg []byte) (router.GossipData, router.GossipData, error) {
var gossip GossipData
if err := gob.NewDecoder(bytes.NewReader(msg)).Decode(&gossip); err != nil {
if err := gossip.Decode(msg); err != nil {
return nil, nil, err
}
if delta := gossip.Timestamp - now(); delta > gossipWindow || delta < -gossipWindow {
return nil, nil, fmt.Errorf("clock skew of %d detected", delta)
}
sort.Sort(CaseInsensitive(gossip.Entries))
n.Lock()
defer n.Unlock()
nameserver/nameserver_test.go
+26 -7
View File
@@ -76,7 +76,7 @@ func (m mapping) Addrs() []address.Address {
}
func TestNameservers(t *testing.T) {
wt.RunWithTimeout(t, time.Minute, func() {
wt.RunWithTimeout(t, 2*time.Minute, func() {
testNameservers(t)
})
}
@@ -85,8 +85,12 @@ func testNameservers(t *testing.T) {
//common.SetLogLevel("debug")
lookupTimeout := 10 // ms
nameservers, grouter := makeNetwork(50)
nameservers, grouter := makeNetwork(30)
defer stopNetwork(nameservers, grouter)
// This subset will sometimes lose touch with some of the others
badNameservers := nameservers[25:]
// This subset will remain well-connected, and we will deal mainly with them
nameservers = nameservers[:25]
nameserversByName := map[router.PeerName]*Nameserver{}
for _, n := range nameservers {
nameserversByName[n.ourName] = n
@@ -110,7 +114,13 @@ func testNameservers(t *testing.T) {
addMapping := func() {
nameserver := nameservers[rand.Intn(len(nameservers))]
addr := address.Address(rand.Int31())
hostname := fmt.Sprintf("hostname%d", rand.Int63())
// Create a hostname which has some upper and lowercase letters,
// and a unique number so we don't have to check if we allocated it already
randomBits := rand.Int63()
firstLetter := 'H' + (randomBits&1)*32
secondLetter := 'O' + (randomBits&2)*16
randomBits = randomBits >> 2
hostname := fmt.Sprintf("%c%cstname%d", firstLetter, secondLetter, randomBits)
mapping := mapping{hostname, []pair{{nameserver.ourName, addr}}}
mappings = append(mappings, mapping)
@@ -133,6 +143,12 @@ func testNameservers(t *testing.T) {
check(nameserver, mapping)
}
loseConnection := func() {
nameserver1 := badNameservers[rand.Intn(len(badNameservers))]
nameserver2 := nameservers[rand.Intn(len(nameservers))]
nameserver1.PeerGone(&router.Peer{Name: nameserver2.ourName})
}
deleteMapping := func() {
if len(mappings) <= 0 {
return
@@ -183,7 +199,7 @@ func testNameservers(t *testing.T) {
require.Equal(t, mapping.hostname, hostname)
}
for i := 0; i < 1000; i++ {
for i := 0; i < 800; i++ {
r := rand.Float32()
switch {
case r < 0.1:
@@ -195,7 +211,10 @@ func testNameservers(t *testing.T) {
case 0.2 <= r && r < 0.3:
deleteMapping()
case 0.3 <= r && r < 0.9:
case 0.3 <= r && r < 0.35:
loseConnection()
case 0.35 <= r && r < 0.9:
doLookup()
case 0.9 <= r:
@@ -245,14 +264,14 @@ func TestTombstoneDeletion(t *testing.T) {
err = nameserver.Delete("hostname", "containerid", "", address.Address(0))
require.Nil(t, err)
require.Equal(t, []address.Address{}, nameserver.Lookup("hostname"))
require.Equal(t, Entries{{
require.Equal(t, l(Entries{Entry{
ContainerID: "containerid",
Origin: peername,
Addr: address.Address(0),
Hostname: "hostname",
Version: 1,
Tombstone: 1234,
}}, nameserver.entries)
}}), nameserver.entries)
now = func() int64 { return 1234 + int64(tombstoneTimeout/time.Second) + 1 }
nameserver.deleteTombstones()
testing/gossip/mocks.go
+1 -1
View File
@@ -97,7 +97,7 @@ type TestRouterClient struct {
}
func (grouter *TestRouter) run(sender router.PeerName, gossiper router.Gossiper, gossipChan chan interface{}) {
gossipTimer := time.Tick(10 * time.Second)
gossipTimer := time.Tick(2 * time.Second)
for {
select {
case gossip := <-gossipChan: