Create a Tracker when starting the router
The tracker's HandleUpdate method is called by the ring update routines. In addition, we add a dummy overlay for AWSVPC to make `weave (status| report)` to produce reasonable reports which includes AWS hosts connection state.
This commit is contained in:
1 parent
c260536429
commit
9fc7be83c8
6 files changed
+206
-55
No files matched your search
+19
-3
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/weaveworks/weave/ipam/paxos"
|
||||
"github.com/weaveworks/weave/ipam/ring"
|
||||
"github.com/weaveworks/weave/ipam/space"
|
||||
"github.com/weaveworks/weave/ipam/tracker"
|
||||
"github.com/weaveworks/weave/net/address"
|
||||
)
|
||||
|
||||
@@ -85,11 +86,14 @@ type Config struct {
|
||||
Quorum func() uint
|
||||
Db db.DB
|
||||
IsKnownPeer func(name mesh.PeerName) bool
|
||||
Tracker tracker.LocalRangeTracker
|
||||
}
|
||||
|
||||
// NewAllocator creates and initialises a new Allocator
|
||||
func NewAllocator(config Config) *Allocator {
|
||||
var participant paxos.Participant
|
||||
var alloc *Allocator
|
||||
var onUpdate ring.OnUpdate
|
||||
|
||||
if config.IsObserver {
|
||||
participant = paxos.NewObserver()
|
||||
@@ -97,11 +101,19 @@ func NewAllocator(config Config) *Allocator {
|
||||
participant = paxos.NewNode(config.OurName, config.OurUID, 0)
|
||||
}
|
||||
|
||||
return &Allocator{
|
||||
if config.Tracker != nil {
|
||||
onUpdate = func(prev []address.Range, curr []address.Range, local bool) {
|
||||
if err := config.Tracker.HandleUpdate(prev, curr, local); err != nil {
|
||||
alloc.errorf("HandleUpdate failed: %s", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
alloc = &Allocator{
|
||||
ourName: config.OurName,
|
||||
seed: config.Seed,
|
||||
universe: config.Universe,
|
||||
ring: ring.New(config.Universe.Range().Start, config.Universe.Range().End, config.OurName),
|
||||
ring: ring.New(config.Universe.Range().Start, config.Universe.Range().End, config.OurName, onUpdate),
|
||||
owned: make(map[string]ownedData),
|
||||
db: config.Db,
|
||||
paxos: participant,
|
||||
@@ -111,6 +123,7 @@ func NewAllocator(config Config) *Allocator {
|
||||
dead: make(map[string]time.Time),
|
||||
now: time.Now,
|
||||
}
|
||||
return alloc
|
||||
}
|
||||
|
||||
func ParseCIDRSubnet(cidrStr string) (cidr address.CIDR, err error) {
|
||||
@@ -958,7 +971,7 @@ func (alloc *Allocator) loadPersistedData() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
alloc.ring = persistedRing
|
||||
alloc.ring.Restore(persistedRing)
|
||||
alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
|
||||
|
||||
if ownedFound {
|
||||
@@ -1068,6 +1081,9 @@ func (alloc *Allocator) fatalf(fmt string, args ...interface{}) {
|
||||
func (alloc *Allocator) warnf(fmt string, args ...interface{}) {
|
||||
alloc.logf(common.Log.Warnf, fmt, args...)
|
||||
}
|
||||
func (alloc *Allocator) errorf(fmt string, args ...interface{}) {
|
||||
common.Log.Errorf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
|
||||
}
|
||||
func (alloc *Allocator) infof(fmt string, args ...interface{}) {
|
||||
alloc.logf(common.Log.Infof, fmt, args...)
|
||||
}
|
||||
|
||||
+35
-3
@@ -18,12 +18,15 @@ import (
|
||||
"github.com/weaveworks/weave/net/address"
|
||||
)
|
||||
|
||||
type OnUpdate func(prev, curr []address.Range, local bool)
|
||||
|
||||
// Ring represents the ring itself
|
||||
type Ring struct {
|
||||
Start, End address.Address // [min, max) tokens in this ring. Due to wrapping, min == max (effectively)
|
||||
Peer mesh.PeerName // name of peer owning this ring instance
|
||||
Entries entries // list of entries sorted by token
|
||||
Seeds []mesh.PeerName // peers with which the ring was seeded
|
||||
onUpdate OnUpdate
|
||||
}
|
||||
|
||||
func (r *Ring) assertInvariants() {
|
||||
@@ -94,15 +97,33 @@ func (r *Ring) checkEntries(entries entries) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Ring) trackUpdates() func() {
|
||||
return r.trackUpdatesOfPeer(r.Peer)
|
||||
}
|
||||
|
||||
func (r *Ring) trackUpdatesOfPeer(peer mesh.PeerName) func() {
|
||||
if r.onUpdate == nil {
|
||||
return func() {}
|
||||
}
|
||||
ranges := r.OwnedRangesOfPeer(peer)
|
||||
return func() { r.onUpdate(ranges, r.OwnedRangesOfPeer(peer), peer == r.Peer) }
|
||||
}
|
||||
|
||||
// New creates an empty ring belonging to peer.
|
||||
func New(start, end address.Address, peer mesh.PeerName) *Ring {
|
||||
func New(start, end address.Address, peer mesh.PeerName, f OnUpdate) *Ring {
|
||||
common.Assert(start < end)
|
||||
|
||||
ring := &Ring{Start: start, End: end, Peer: peer, Entries: make([]*entry, 0)}
|
||||
ring := &Ring{Start: start, End: end, Peer: peer, Entries: make([]*entry, 0), onUpdate: f}
|
||||
ring.updateExportedVariables()
|
||||
return ring
|
||||
}
|
||||
|
||||
func (r *Ring) Restore(other *Ring) {
|
||||
onUpdate := r.onUpdate
|
||||
*r = *other
|
||||
r.onUpdate = onUpdate
|
||||
}
|
||||
|
||||
func (r *Ring) Range() address.Range {
|
||||
return address.Range{Start: r.Start, End: r.End}
|
||||
}
|
||||
@@ -126,6 +147,7 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
|
||||
//fmt.Printf("%s GrantRangeToHost [%v,%v) -> %s\n", r.Peer, start, end, peer)
|
||||
|
||||
r.assertInvariants()
|
||||
defer r.trackUpdates()()
|
||||
defer r.assertInvariants()
|
||||
defer r.updateExportedVariables()
|
||||
|
||||
@@ -193,6 +215,7 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
|
||||
// got updated as a result.
|
||||
func (r *Ring) Merge(gossip Ring) (bool, error) {
|
||||
r.assertInvariants()
|
||||
defer r.trackUpdates()()
|
||||
defer r.updateExportedVariables()
|
||||
|
||||
// Don't panic when checking the gossiped in ring.
|
||||
@@ -230,6 +253,7 @@ func (r *Ring) Merge(gossip Ring) (bool, error) {
|
||||
r.Seeds = gossip.Seeds
|
||||
}
|
||||
r.Entries = result
|
||||
|
||||
return updated, nil
|
||||
}
|
||||
|
||||
@@ -336,10 +360,14 @@ func (r *Ring) splitRangesOverZero(ranges []address.Range) []address.Range {
|
||||
// ranges are owned by this peer. Will split ranges which
|
||||
// span 0 in the ring.
|
||||
func (r *Ring) OwnedRanges() (result []address.Range) {
|
||||
return r.OwnedRangesOfPeer(r.Peer)
|
||||
}
|
||||
|
||||
func (r *Ring) OwnedRangesOfPeer(peer mesh.PeerName) (result []address.Range) {
|
||||
r.assertInvariants()
|
||||
|
||||
for i, entry := range r.Entries {
|
||||
if entry.Peer == r.Peer {
|
||||
if entry.Peer == peer {
|
||||
nextEntry := r.Entries.entry(i + 1)
|
||||
result = append(result, address.Range{Start: entry.Token, End: nextEntry.Token})
|
||||
}
|
||||
@@ -371,6 +399,8 @@ func (r *Ring) AllRangeInfo() (result []RangeInfo) {
|
||||
// in. Only works for empty rings. Each claimed range is CIDR-aligned.
|
||||
func (r *Ring) ClaimForPeers(peers []mesh.PeerName) {
|
||||
common.Assert(r.Empty())
|
||||
|
||||
defer r.trackUpdates()()
|
||||
defer r.assertInvariants()
|
||||
defer r.updateExportedVariables()
|
||||
defer func() {
|
||||
@@ -527,6 +557,8 @@ func (r *Ring) PickPeerForTransfer(isValid func(mesh.PeerName) bool) mesh.PeerNa
|
||||
// and return ranges indicating the new space we picked up
|
||||
func (r *Ring) Transfer(from, to mesh.PeerName) []address.Range {
|
||||
r.assertInvariants()
|
||||
defer r.trackUpdates()()
|
||||
defer r.trackUpdatesOfPeer(from)()
|
||||
defer r.assertInvariants()
|
||||
defer r.updateExportedVariables()
|
||||
|
||||
|
||||
+45
-41
@@ -36,8 +36,12 @@ func merge(r1, r2 *Ring) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func NewRing(start, end address.Address, peer mesh.PeerName) *Ring {
|
||||
return New(start, end, peer, nil)
|
||||
}
|
||||
|
||||
func TestInvariants(t *testing.T) {
|
||||
ring := New(start, end, peer1name)
|
||||
ring := NewRing(start, end, peer1name)
|
||||
|
||||
// Check ring is sorted
|
||||
ring.Entries = []*entry{{Token: dot245, Peer: peer1name}, {Token: dot10, Peer: peer2name}}
|
||||
@@ -48,7 +52,7 @@ func TestInvariants(t *testing.T) {
|
||||
require.True(t, ring.checkInvariants() == ErrTokenRepeated, "Expected error")
|
||||
|
||||
// Check tokens are in bounds
|
||||
ring = New(dot10, dot245, peer1name)
|
||||
ring = NewRing(dot10, dot245, peer1name)
|
||||
ring.Entries = []*entry{{Token: start, Peer: peer1name}}
|
||||
require.True(t, ring.checkInvariants() == ErrTokenOutOfRange, "Expected error")
|
||||
|
||||
@@ -57,7 +61,7 @@ func TestInvariants(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestInsert(t *testing.T) {
|
||||
ring := New(start, end, peer1name)
|
||||
ring := NewRing(start, end, peer1name)
|
||||
ring.Entries = []*entry{{Token: start, Peer: peer1name, Free: 255}}
|
||||
|
||||
require.Panics(t, func() {
|
||||
@@ -82,7 +86,7 @@ func TestInsert(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestBetween(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring1.Entries = []*entry{{Token: start, Peer: peer1name, Free: 255}}
|
||||
|
||||
// First off, in a ring where everything is owned by the peer
|
||||
@@ -124,8 +128,8 @@ func TestBetween(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGrantSimple(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
// Claim everything for peer1
|
||||
ring1.ClaimItAll()
|
||||
@@ -159,8 +163,8 @@ func TestGrantSimple(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGrantSplit(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
// Claim everything for peer1
|
||||
ring1.Entries = []*entry{{Token: start, Peer: peer1name, Free: 256}}
|
||||
@@ -188,8 +192,8 @@ func TestGrantSplit(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMergeSimple(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
// Claim everything for peer1
|
||||
ring1.ClaimItAll()
|
||||
@@ -217,18 +221,18 @@ func TestMergeSimple(t *testing.T) {
|
||||
|
||||
func TestMergeErrors(t *testing.T) {
|
||||
// Cannot Merge in an invalid ring
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
ring2.Entries = []*entry{{Token: middle, Peer: peer2name}, {Token: start, Peer: peer2name}}
|
||||
require.True(t, merge(ring1, ring2) == ErrNotSorted, "Expected ErrNotSorted")
|
||||
|
||||
// Should Merge two rings for different ranges
|
||||
ring2 = New(start, middle, peer2name)
|
||||
ring2 = NewRing(start, middle, peer2name)
|
||||
ring2.Entries = []*entry{}
|
||||
require.True(t, merge(ring1, ring2) == ErrDifferentRange, "Expected ErrDifferentRange")
|
||||
|
||||
// Cannot Merge newer version of entry I own
|
||||
ring2 = New(start, end, peer2name)
|
||||
ring2 = NewRing(start, end, peer2name)
|
||||
ring1.Entries = []*entry{{Token: start, Peer: peer1name}}
|
||||
ring2.Entries = []*entry{{Token: start, Peer: peer1name, Version: 1}}
|
||||
fmt.Println(merge(ring1, ring2))
|
||||
@@ -246,8 +250,8 @@ func TestMergeErrors(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMergeMore(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
assertRing := func(ring *Ring, entries entries) {
|
||||
require.Equal(t, entries, ring.Entries)
|
||||
@@ -297,8 +301,8 @@ func TestMergeMore(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMergeSplit(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
// Claim everything for peer2
|
||||
ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 256}}
|
||||
@@ -320,8 +324,8 @@ func TestMergeSplit(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMergeSplit2(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
// Claim everything for peer2
|
||||
ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 250}, {Token: dot250, Peer: peer2name, Free: 5}}
|
||||
@@ -344,8 +348,8 @@ func TestMergeSplit2(t *testing.T) {
|
||||
|
||||
// A simple test, very similar to above, but using the marshalling to byte[]s
|
||||
func TestGossip(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
assertRing := func(ring *Ring, entries entries) {
|
||||
require.Equal(t, entries, ring.Entries)
|
||||
@@ -372,7 +376,7 @@ func assertPeersWithSpace(t *testing.T, ring *Ring, start, end address.Address,
|
||||
}
|
||||
|
||||
func TestFindFree(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
|
||||
assertPeersWithSpace(t, ring1, start, end, 0)
|
||||
|
||||
@@ -408,8 +412,8 @@ func TestFindFree(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestReportFree(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
ring1.ClaimItAll()
|
||||
ring1.GrantRangeToHost(middle, end, peer2name)
|
||||
@@ -423,7 +427,7 @@ func TestReportFree(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMisc(t *testing.T) {
|
||||
ring := New(start, end, peer1name)
|
||||
ring := NewRing(start, end, peer1name)
|
||||
|
||||
require.True(t, ring.Empty(), "empty")
|
||||
|
||||
@@ -432,8 +436,8 @@ func TestMisc(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestEmptyGossip(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
ring1.ClaimItAll()
|
||||
// This used to panic, and it shouldn't
|
||||
@@ -441,8 +445,8 @@ func TestEmptyGossip(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMergeOldMessage(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
ring1.ClaimItAll()
|
||||
require.NoError(t, merge(ring2, ring1))
|
||||
@@ -452,8 +456,8 @@ func TestMergeOldMessage(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSplitRangeAtBeginning(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
|
||||
ring1.ClaimItAll()
|
||||
require.NoError(t, merge(ring2, ring1))
|
||||
@@ -463,7 +467,7 @@ func TestSplitRangeAtBeginning(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestOwnedRange(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring1.ClaimItAll()
|
||||
|
||||
require.Equal(t, []address.Range{{Start: start, End: end}}, ring1.OwnedRanges())
|
||||
@@ -471,7 +475,7 @@ func TestOwnedRange(t *testing.T) {
|
||||
ring1.GrantRangeToHost(middle, end, peer2name)
|
||||
require.Equal(t, []address.Range{{Start: start, End: middle}}, ring1.OwnedRanges())
|
||||
|
||||
ring2 := New(start, end, peer2name)
|
||||
ring2 := NewRing(start, end, peer2name)
|
||||
merge(ring2, ring1)
|
||||
require.Equal(t, []address.Range{{Start: middle, End: end}}, ring2.OwnedRanges())
|
||||
|
||||
@@ -486,7 +490,7 @@ func TestOwnedRange(t *testing.T) {
|
||||
|
||||
func TestTransfer(t *testing.T) {
|
||||
// First test just checks if we can grant some range to a host, when we transfer it, we get it back
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
ring1.ClaimItAll()
|
||||
ring1.GrantRangeToHost(middle, end, peer2name)
|
||||
ring1.Transfer(peer2name, peer1name)
|
||||
@@ -494,7 +498,7 @@ func TestTransfer(t *testing.T) {
|
||||
|
||||
// Second test is what happens when a token exists at the end of a range but is transferred
|
||||
// - does it get resurrected correctly?
|
||||
ring1 = New(start, end, peer1name)
|
||||
ring1 = NewRing(start, end, peer1name)
|
||||
ring1.ClaimItAll()
|
||||
ring1.GrantRangeToHost(middle, end, peer2name)
|
||||
ring1.Transfer(peer2name, peer1name)
|
||||
@@ -503,7 +507,7 @@ func TestTransfer(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestOwner(t *testing.T) {
|
||||
ring1 := New(start, end, peer1name)
|
||||
ring1 := NewRing(start, end, peer1name)
|
||||
require.True(t, ring1.Contains(start), "start should be in ring")
|
||||
require.False(t, ring1.Contains(end), "end should not be in ring")
|
||||
|
||||
@@ -544,7 +548,7 @@ func TestClaimForPeers(t *testing.T) {
|
||||
peers := makePeers(numPeers)
|
||||
// Test for a range of peer counts
|
||||
for i := 0; i < numPeers; i++ {
|
||||
ring := New(start, end, peers[0])
|
||||
ring := NewRing(start, end, peers[0])
|
||||
ring.ClaimForPeers(peers[:i+1])
|
||||
}
|
||||
}
|
||||
@@ -582,7 +586,7 @@ func TestFuzzRing(t *testing.T) {
|
||||
sort.Sort(addressSlice(tokens))
|
||||
|
||||
peer := peers[rand.Intn(len(peers))]
|
||||
ring := New(start, end, peer)
|
||||
ring := NewRing(start, end, peer)
|
||||
for _, token := range tokens {
|
||||
peer = peers[rand.Intn(len(peers))]
|
||||
ring.Entries = append(ring.Entries, &entry{Token: start + token, Peer: peer})
|
||||
@@ -616,7 +620,7 @@ func TestFuzzRing(t *testing.T) {
|
||||
}
|
||||
|
||||
peer := peers[rand.Intn(len(peers))]
|
||||
ring := New(start, end, peer)
|
||||
ring := NewRing(start, end, peer)
|
||||
for _, token := range tokens {
|
||||
peer = peers[rand.Intn(len(peers))]
|
||||
ring.Entries = append(ring.Entries, &entry{Token: start + token, Peer: peer})
|
||||
@@ -654,7 +658,7 @@ func TestFuzzRingHard(t *testing.T) {
|
||||
common.Log.Debugf("%s: Adding peer", peer)
|
||||
nextPeerID++
|
||||
peers = append(peers, peer)
|
||||
rings = append(rings, New(start, end, peer))
|
||||
rings = append(rings, NewRing(start, end, peer))
|
||||
}
|
||||
|
||||
for i := 0; i < numPeers; i++ {
|
||||
|
||||
+31
-7
@@ -22,6 +22,7 @@ import (
|
||||
"github.com/weaveworks/weave/common/docker"
|
||||
"github.com/weaveworks/weave/db"
|
||||
"github.com/weaveworks/weave/ipam"
|
||||
"github.com/weaveworks/weave/ipam/tracker"
|
||||
"github.com/weaveworks/weave/nameserver"
|
||||
weavenet "github.com/weaveworks/weave/net"
|
||||
"github.com/weaveworks/weave/net/address"
|
||||
@@ -170,6 +171,7 @@ func main() {
|
||||
datapathName string
|
||||
trustedSubnetStr string
|
||||
dbPrefix string
|
||||
isAWSVPC bool
|
||||
|
||||
defaultDockerHost = "unix:///var/run/docker.sock"
|
||||
)
|
||||
@@ -208,6 +210,7 @@ func main() {
|
||||
mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name")
|
||||
mflag.StringVar(&trustedSubnetStr, []string{"-trusted-subnets"}, "", "comma-separated list of trusted subnets in CIDR notation")
|
||||
mflag.StringVar(&dbPrefix, []string{"-db-prefix"}, "/weavedb/weave", "pathname/prefix of filename to store data")
|
||||
mflag.BoolVar(&isAWSVPC, []string{"#awsvpc", "-awsvpc"}, false, "use AWS VPC for routing")
|
||||
|
||||
// crude way of detecting that we probably have been started in a
|
||||
// container, with `weave launch` --> suppress misleading paths in
|
||||
@@ -250,7 +253,7 @@ func main() {
|
||||
networkConfig.PacketLogging = nopPacketLogging{}
|
||||
}
|
||||
|
||||
overlay, bridge := createOverlay(datapathName, ifaceName, config.Host, config.Port, bufSzMB)
|
||||
overlay, bridge := createOverlay(datapathName, ifaceName, isAWSVPC, config.Host, config.Port, bufSzMB)
|
||||
networkConfig.Bridge = bridge
|
||||
|
||||
name := peerName(routerName, bridge.Interface())
|
||||
@@ -302,7 +305,15 @@ func main() {
|
||||
defaultSubnet address.CIDR
|
||||
)
|
||||
if ipamConfig.Enabled() {
|
||||
allocator, defaultSubnet = createAllocator(router, ipamConfig, db, isKnownPeer)
|
||||
var t tracker.LocalRangeTracker
|
||||
if isAWSVPC {
|
||||
Log.Infoln("Creating AWSVPC LocalRangeTracker")
|
||||
t, err = tracker.NewAWSVPCTracker()
|
||||
if err != nil {
|
||||
Log.Fatalf("Cannot create AWSVPC LocalRangeTracker: %s", err)
|
||||
}
|
||||
}
|
||||
allocator, defaultSubnet = createAllocator(router, ipamConfig, db, t, isKnownPeer)
|
||||
observeContainers(allocator)
|
||||
ids, err := dockerCli.AllContainerIDs()
|
||||
checkFatal(err)
|
||||
@@ -388,10 +399,18 @@ func (nopPacketLogging) LogPacket(string, weave.PacketKey) {
|
||||
func (nopPacketLogging) LogForwardPacket(string, weave.ForwardPacketKey) {
|
||||
}
|
||||
|
||||
func createOverlay(datapathName string, ifaceName string, host string, port int, bufSzMB int) (weave.NetworkOverlay, weave.Bridge) {
|
||||
func createOverlay(datapathName string, ifaceName string, isAWSVPC bool, host string, port int, bufSzMB int) (weave.NetworkOverlay, weave.Bridge) {
|
||||
overlay := weave.NewOverlaySwitch()
|
||||
var bridge weave.Bridge
|
||||
var ignoreSleeve bool
|
||||
|
||||
switch {
|
||||
case isAWSVPC:
|
||||
vpc := weave.NewAWSVPC()
|
||||
overlay.Add("awsvpc", vpc)
|
||||
bridge = weave.NullBridge{}
|
||||
// Currently, we do not support any overlay with AWSVPC
|
||||
ignoreSleeve = true
|
||||
case datapathName != "" && ifaceName != "":
|
||||
Log.Fatal("At most one of --datapath and --iface must be specified.")
|
||||
case datapathName != "":
|
||||
@@ -409,13 +428,17 @@ func createOverlay(datapathName string, ifaceName string, host string, port int,
|
||||
default:
|
||||
bridge = weave.NullBridge{}
|
||||
}
|
||||
sleeve := weave.NewSleeveOverlay(host, port)
|
||||
overlay.Add("sleeve", sleeve)
|
||||
overlay.SetCompatOverlay(sleeve)
|
||||
|
||||
if !ignoreSleeve {
|
||||
sleeve := weave.NewSleeveOverlay(host, port)
|
||||
overlay.Add("sleeve", sleeve)
|
||||
overlay.SetCompatOverlay(sleeve)
|
||||
}
|
||||
|
||||
return overlay, bridge
|
||||
}
|
||||
|
||||
func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
|
||||
func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, track tracker.LocalRangeTracker, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
|
||||
ipRange, err := ipam.ParseCIDRSubnet(config.IPRangeCIDR)
|
||||
checkFatal(err)
|
||||
defaultSubnet := ipRange
|
||||
@@ -437,6 +460,7 @@ func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, i
|
||||
Quorum: func() uint { return determineQuorum(config.PeerCount, router) },
|
||||
Db: db,
|
||||
IsKnownPeer: isKnownPeer,
|
||||
Tracker: track,
|
||||
}
|
||||
|
||||
allocator := ipam.NewAllocator(c)
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
package router
|
||||
|
||||
// A dummy overlay for the AWSVPC underlay to make `weave status` to return
|
||||
// a valid information about peer connections.
|
||||
|
||||
import (
|
||||
"github.com/weaveworks/mesh"
|
||||
)
|
||||
|
||||
// mesh.OverlayConnection
|
||||
|
||||
type AWSVPCConnection struct {
|
||||
establishedChan chan struct{}
|
||||
errorChan chan error
|
||||
}
|
||||
|
||||
func (conn *AWSVPCConnection) Confirm() {
|
||||
// We close the channel to notify mesh that the connection has been established.
|
||||
close(conn.establishedChan)
|
||||
}
|
||||
|
||||
func (conn *AWSVPCConnection) EstablishedChannel() <-chan struct{} {
|
||||
return conn.establishedChan
|
||||
}
|
||||
|
||||
func (conn *AWSVPCConnection) ErrorChannel() <-chan error {
|
||||
return conn.errorChan
|
||||
}
|
||||
|
||||
func (conn *AWSVPCConnection) Stop() {}
|
||||
|
||||
func (conn *AWSVPCConnection) ControlMessage(tag byte, msg []byte) {
|
||||
}
|
||||
|
||||
func (conn *AWSVPCConnection) DisplayName() string {
|
||||
return "awsvpc"
|
||||
}
|
||||
|
||||
// OverlayForwarder
|
||||
|
||||
func (conn *AWSVPCConnection) Forward(key ForwardPacketKey) FlowOp {
|
||||
return DiscardingFlowOp{}
|
||||
}
|
||||
|
||||
type AWSVPC struct{}
|
||||
|
||||
func NewAWSVPC() AWSVPC {
|
||||
return AWSVPC{}
|
||||
}
|
||||
|
||||
// mesh.Overlay
|
||||
|
||||
func (vpc AWSVPC) AddFeaturesTo(features map[string]string) {}
|
||||
|
||||
func (vpc AWSVPC) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) {
|
||||
conn := &AWSVPCConnection{
|
||||
establishedChan: make(chan struct{}),
|
||||
errorChan: make(chan error, 1),
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (vpc AWSVPC) Diagnostics() interface{} {
|
||||
return nil
|
||||
}
|
||||
|
||||
// NetworkOverlay
|
||||
|
||||
func (vpc AWSVPC) InvalidateRoutes() {}
|
||||
|
||||
func (vpc AWSVPC) InvalidateShortIDs() {}
|
||||
|
||||
func (vpc AWSVPC) StartConsumingPackets(localPeer *mesh.Peer, peers *mesh.Peers, consumer OverlayConsumer) error {
|
||||
return nil
|
||||
}
|
||||
+1
-1
@@ -36,7 +36,7 @@ func (NullBridge) Interface() *net.Interface {
|
||||
}
|
||||
|
||||
func (NullBridge) String() string {
|
||||
return "<no bridge networking>"
|
||||
return "no overlay bridge"
|
||||
}
|
||||
|
||||
func (NullBridge) Stats() map[string]int {
|
||||
|
||||
Reference in new issue
Block a user