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

Store address+subnet prefix for 'owned' addresses

Browse Source
This commit is contained in:
Bryan Boreham committed 2016-03-02 17:21:24 -08:00
1 parent b6eec08247
commit b76e2a87cf
7 files changed
+100 -104
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ipam/allocate.go
+2 -2
View File
@@ -28,7 +28,7 @@ func (g *allocate) Try(alloc *Allocator) bool {
if addrs := alloc.ownedInRange(g.ident, g.r); len(addrs) > 0 {
// If we had heard that this container died, resurrect it
delete(alloc.dead, g.ident) // delete is no-op if key not in map
g.resultChan <- allocateResult{addrs[0], nil}
g.resultChan <- allocateResult{addrs[0].Addr, nil}
return true
}
@@ -46,7 +46,7 @@ func (g *allocate) Try(alloc *Allocator) bool {
g.ident = addr.String()
}
alloc.debugln("Allocated", addr, "for", g.ident, "in", g.r)
alloc.addOwned(g.ident, addr)
alloc.addOwned(g.ident, address.MakeCIDR(g.r, addr))
g.resultChan <- allocateResult{addr, nil}
return true
}
ipam/allocator.go
+43 -47
View File
@@ -47,16 +47,16 @@ type operation interface {
type Allocator struct {
actionChan chan<- func()
ourName mesh.PeerName
universe address.Range // superset of all ranges
ring *ring.Ring // information on ranges owned by all peers
space space.Space // more detail on ranges owned by us
owned map[string][]address.Address // who owns what addresses, indexed by container-ID
nicknames map[mesh.PeerName]string // so we can map nicknames for rmpeer
pendingAllocates []operation // held until we get some free space
pendingClaims []operation // held until we know who owns the space
dead map[string]time.Time // containers we heard were dead, and when
db db.DB // persistence
gossip mesh.Gossip // our link to the outside world for sending messages
universe address.Range // superset of all ranges
ring *ring.Ring // information on ranges owned by all peers
space space.Space // more detail on ranges owned by us
owned map[string][]address.CIDR // who owns what addresses, indexed by container-ID
nicknames map[mesh.PeerName]string // so we can map nicknames for rmpeer
pendingAllocates []operation // held until we get some free space
pendingClaims []operation // held until we know who owns the space
dead map[string]time.Time // containers we heard were dead, and when
db db.DB // persistence
gossip mesh.Gossip // our link to the outside world for sending messages
paxos *paxos.Node
paxosActive bool
ticker *time.Ticker
@@ -71,7 +71,7 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string
ourName: ourName,
universe: universe,
ring: ring.New(universe.Start, universe.End, ourName),
owned: make(map[string][]address.Address),
owned: make(map[string][]address.CIDR),
db: db,
paxos: paxos.NewNode(ourName, ourUID, quorum),
nicknames: map[mesh.PeerName]string{ourName: ourNickname},
@@ -173,7 +173,7 @@ func (alloc *Allocator) tryPendingOps() {
func (alloc *Allocator) spaceRequestDenied(sender mesh.PeerName, r address.Range) {
for i := 0; i < len(alloc.pendingClaims); {
claim := alloc.pendingClaims[i].(*claim)
if r.Contains(claim.addr) {
if r.Contains(claim.cidr.Addr) {
claim.deniedBy(alloc, sender)
alloc.pendingClaims = append(alloc.pendingClaims[:i], alloc.pendingClaims[i+1:]...)
continue
@@ -204,8 +204,8 @@ func (alloc *Allocator) Allocate(ident string, r address.Range, hasBeenCancelled
}
// Lookup (Sync) - get existing IP addresses for container with given name in range
func (alloc *Allocator) Lookup(ident string, r address.Range) ([]address.Address, error) {
resultChan := make(chan []address.Address)
func (alloc *Allocator) Lookup(ident string, r address.Range) ([]address.CIDR, error) {
resultChan := make(chan []address.CIDR)
alloc.actionChan <- func() {
resultChan <- alloc.ownedInRange(ident, r)
}
@@ -213,9 +213,9 @@ func (alloc *Allocator) Lookup(ident string, r address.Range) ([]address.Address
}
// Claim an address that we think we should own (Sync)
func (alloc *Allocator) Claim(ident string, addr address.Address, noErrorOnUnknown bool) error {
func (alloc *Allocator) Claim(ident string, cidr address.CIDR, noErrorOnUnknown bool) error {
resultChan := make(chan error)
op := &claim{resultChan: resultChan, ident: ident, addr: addr, noErrorOnUnknown: noErrorOnUnknown}
op := &claim{resultChan: resultChan, ident: ident, cidr: cidr, noErrorOnUnknown: noErrorOnUnknown}
alloc.doOperation(op, &alloc.pendingClaims)
return <-resultChan
}
@@ -282,12 +282,12 @@ func (alloc *Allocator) Delete(ident string) error {
}
func (alloc *Allocator) delete(ident string) error {
addrs := alloc.removeAllOwned(ident)
if len(addrs) == 0 {
cidrs := alloc.removeAllOwned(ident)
if len(cidrs) == 0 {
return fmt.Errorf("Delete: no addresses for %s", ident)
}
for _, addr := range addrs {
alloc.space.Free(addr)
for _, cidr := range cidrs {
alloc.space.Free(cidr.Addr)
}
return nil
}
@@ -803,9 +803,9 @@ func (alloc *Allocator) loadPersistedData() {
if err := alloc.db.Load(ownedIdent, &alloc.owned); err != nil {
alloc.fatalf("Error loading persisted address data: %s", err)
}
for _, addrs := range alloc.owned {
for _, addr := range addrs {
alloc.space.Claim(addr)
for _, cidrs := range alloc.owned {
for _, cidr := range cidrs {
alloc.space.Claim(cidr.Addr)
}
}
}
@@ -818,22 +818,18 @@ func (alloc *Allocator) persistOwned() {
// Owned addresses
func (alloc *Allocator) allOwned(ident string) []address.Address {
return alloc.owned[ident]
}
func (alloc *Allocator) hasOwned(ident string) bool {
_, b := alloc.owned[ident]
return b
}
// NB: addr must not be owned by ident already
func (alloc *Allocator) addOwned(ident string, addr address.Address) {
alloc.owned[ident] = append(alloc.owned[ident], addr)
func (alloc *Allocator) addOwned(ident string, cidr address.CIDR) {
alloc.owned[ident] = append(alloc.owned[ident], cidr)
alloc.persistOwned()
}
func (alloc *Allocator) removeAllOwned(ident string) []address.Address {
func (alloc *Allocator) removeAllOwned(ident string) []address.CIDR {
a := alloc.owned[ident]
delete(alloc.owned, ident)
alloc.persistOwned()
@@ -841,13 +837,13 @@ func (alloc *Allocator) removeAllOwned(ident string) []address.Address {
}
func (alloc *Allocator) removeOwned(ident string, addrToFree address.Address) bool {
addrs, _ := alloc.owned[ident]
for i, ownedAddr := range addrs {
if ownedAddr == addrToFree {
if len(addrs) == 1 {
cidrs, _ := alloc.owned[ident]
for i, ownedCidr := range cidrs {
if ownedCidr.Addr == addrToFree {
if len(cidrs) == 1 {
delete(alloc.owned, ident)
} else {
alloc.owned[ident] = append(addrs[:i], addrs[i+1:]...)
alloc.owned[ident] = append(cidrs[:i], cidrs[i+1:]...)
}
alloc.persistOwned()
return true
@@ -856,20 +852,20 @@ func (alloc *Allocator) removeOwned(ident string, addrToFree address.Address) bo
return false
}
func (alloc *Allocator) ownedInRange(ident string, r address.Range) []address.Address {
var a []address.Address
for _, addr := range alloc.owned[ident] {
if r.Contains(addr) {
a = append(a, addr)
func (alloc *Allocator) ownedInRange(ident string, r address.Range) []address.CIDR {
var c []address.CIDR
for _, cidr := range alloc.owned[ident] {
if r.Contains(cidr.Addr) {
c = append(c, cidr)
}
}
return a
return c
}
func (alloc *Allocator) findOwner(addr address.Address) string {
for ident, addrs := range alloc.owned {
for _, candidate := range addrs {
if candidate == addr {
for ident, cidrs := range alloc.owned {
for _, candidate := range cidrs {
if candidate.Addr == addr {
return ident
}
}
@@ -880,10 +876,10 @@ func (alloc *Allocator) findOwner(addr address.Address) string {
// For each ID in the 'owned' map, remove the entry if it isn't in the map
func (alloc *Allocator) syncOwned(ids map[string]struct{}) {
changed := false
for ident, addrs := range alloc.owned {
for ident, cidrs := range alloc.owned {
if _, found := ids[ident]; !found {
for _, addr := range addrs {
alloc.space.Free(addr)
for _, cidr := range cidrs {
alloc.space.Free(cidr.Addr)
}
delete(alloc.owned, ident)
changed = true
ipam/claim.go
+20 -20
View File
@@ -12,7 +12,7 @@ import (
type claim struct {
resultChan chan<- error
ident string
addr address.Address
cidr address.CIDR
noErrorOnUnknown bool
}
@@ -32,10 +32,10 @@ func (c *claim) sendResult(result error) {
// Try returns true for success (or failure), false if we need to try again later
func (c *claim) Try(alloc *Allocator) bool {
if !alloc.ring.Contains(c.addr) {
if !alloc.ring.Contains(c.cidr.Addr) {
// Address not within our universe; assume user knows what they are doing
alloc.infof("Address %s claimed by %s - not in our range", c.addr, c.ident)
alloc.addOwned(c.ident, c.addr)
alloc.infof("Address %s claimed by %s - not in our range", c.cidr, c.ident)
alloc.addOwned(c.ident, c.cidr)
c.sendResult(nil)
return true
}
@@ -45,24 +45,24 @@ func (c *claim) Try(alloc *Allocator) bool {
// If we had heard that this container died, resurrect it
delete(alloc.dead, c.ident) // (delete is no-op if key not in map)
switch owner := alloc.ring.Owner(c.addr); owner {
switch owner := alloc.ring.Owner(c.cidr.Addr); owner {
case alloc.ourName:
// success
case mesh.UnknownPeerName:
// If our ring doesn't know, it must be empty.
if c.noErrorOnUnknown {
alloc.infof("Claim %s for %s: address allocator still initializing; will try later.", c.addr, c.ident)
alloc.infof("Claim %s for %s: address allocator still initializing; will try later.", c.cidr, c.ident)
c.sendResult(nil)
} else {
c.sendResult(fmt.Errorf("%s is in the range %s, but the allocator is not initialized yet", c.addr, alloc.universe.AsCIDRString()))
c.sendResult(fmt.Errorf("%s is in the range %s, but the allocator is not initialized yet", c.cidr, alloc.universe.AsCIDRString()))
}
return false
default:
alloc.debugf("requesting address %s from other peer %s", c.addr, owner)
err := alloc.sendSpaceRequest(owner, address.NewRange(c.addr, 1))
alloc.debugf("requesting address %s from other peer %s", c.cidr, owner)
err := alloc.sendSpaceRequest(owner, address.NewRange(c.cidr.Addr, 1))
if err != nil { // can't speak to owner right now
if c.noErrorOnUnknown {
alloc.infof("Claim %s for %s: %s; will try later.", c.addr, c.ident, err)
alloc.infof("Claim %s for %s: %s; will try later.", c.cidr, c.ident, err)
c.sendResult(nil)
} else { // just tell the user they can't do this.
c.deniedBy(alloc, owner)
@@ -72,28 +72,28 @@ func (c *claim) Try(alloc *Allocator) bool {
}
// We are the owner, check we haven't given it to another container
switch existingIdent := alloc.findOwner(c.addr); existingIdent {
switch existingIdent := alloc.findOwner(c.cidr.Addr); existingIdent {
case "":
if err := alloc.space.Claim(c.addr); err == nil {
alloc.debugln("Claimed", c.addr, "for", c.ident)
if err := alloc.space.Claim(c.cidr.Addr); err == nil {
alloc.debugln("Claimed", c.cidr, "for", c.ident)
if c.ident == "_" { // Special "I don't have a unique ID" identifier
c.ident = c.addr.String()
c.ident = c.cidr.String()
}
alloc.addOwned(c.ident, c.addr)
alloc.addOwned(c.ident, c.cidr)
c.sendResult(nil)
} else {
c.sendResult(err)
}
case c.ident:
// same identifier is claiming same address; that's OK
alloc.debugln("Re-Claimed", c.addr, "for", c.ident)
alloc.debugln("Re-Claimed", c.cidr, "for", c.ident)
c.sendResult(nil)
case c.addr.String():
case c.cidr.String():
// Address already allocated via "_" name
c.sendResult(fmt.Errorf("address %s already in use", c.addr))
c.sendResult(fmt.Errorf("address %s already in use", c.cidr))
default:
// Addr already owned by container on this machine
c.sendResult(fmt.Errorf("address %s is already owned by %s", c.addr.String(), existingIdent))
c.sendResult(fmt.Errorf("address %s is already owned by %s", c.cidr.String(), existingIdent))
}
return true
}
@@ -103,7 +103,7 @@ func (c *claim) deniedBy(alloc *Allocator, owner mesh.PeerName) {
if found {
name = " (" + name + ")"
}
c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name))
c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.cidr.String(), owner, name))
}
func (c *claim) Cancel() {
ipam/http.go
+19 -23
View File
@@ -16,23 +16,23 @@ func badRequest(w http.ResponseWriter, err error) {
common.Log.Warningln("[allocator]:", err.Error())
}
func parseCIDR(w http.ResponseWriter, cidrStr string) (address.CIDR, bool) {
func parseCIDR(w http.ResponseWriter, cidrStr string, net bool) (address.CIDR, bool) {
subnetAddr, cidr, err := address.ParseCIDR(cidrStr)
if err != nil {
badRequest(w, err)
return address.CIDR{}, false
}
if cidr.Start != subnetAddr {
if net && cidr.Addr != subnetAddr {
badRequest(w, fmt.Errorf("Invalid subnet %s - bits after network prefix are not all zero", cidrStr))
return address.CIDR{}, false
}
return cidr, true
}
func writeAddresses(w http.ResponseWriter, addrs []address.Address, subnet address.CIDR) {
for i, addr := range addrs {
fmt.Fprintf(w, "%s/%d", addr, subnet.PrefixLen)
if i < len(addrs)-1 {
func writeAddresses(w http.ResponseWriter, cidrs []address.CIDR) {
for i, cidr := range cidrs {
fmt.Fprint(w, cidr)
if i < len(cidrs)-1 {
w.Write([]byte{' '})
}
}
@@ -70,31 +70,27 @@ func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CID
fmt.Fprintf(w, "%s", defaultSubnet)
})
router.Methods("PUT").Path("/ip/{id}/{ip}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
router.Methods("PUT").Path("/ip/{id}/{ip}/{prefixlen}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
ident := vars["id"]
ipStr := vars["ip"]
noErrorOnUnknown := r.FormValue("noErrorOnUnknown") == "true"
if ip, err := address.ParseIP(ipStr); err != nil {
badRequest(w, err)
return
} else if err := alloc.Claim(ident, ip, noErrorOnUnknown); err != nil {
badRequest(w, fmt.Errorf("Unable to claim: %s", err))
return
if cidr, ok := parseCIDR(w, vars["ip"]+"/"+vars["prefixlen"], false); ok {
noErrorOnUnknown := r.FormValue("noErrorOnUnknown") == "true"
if err := alloc.Claim(vars["id"], cidr, noErrorOnUnknown); err != nil {
badRequest(w, fmt.Errorf("Unable to claim: %s", err))
return
}
w.WriteHeader(204)
}
w.WriteHeader(204)
})
router.Methods("GET").Path("/ip/{id}/{ip}/{prefixlen}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
if subnet, ok := parseCIDR(w, vars["ip"]+"/"+vars["prefixlen"]); ok {
addrs, err := alloc.Lookup(vars["id"], subnet.HostRange())
if subnet, ok := parseCIDR(w, vars["ip"]+"/"+vars["prefixlen"], true); ok {
cidrs, err := alloc.Lookup(vars["id"], subnet.HostRange())
if err != nil {
http.NotFound(w, r)
return
}
writeAddresses(w, addrs, subnet)
writeAddresses(w, cidrs)
}
})
@@ -104,12 +100,12 @@ func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CID
http.NotFound(w, r)
return
}
writeAddresses(w, addrs, defaultSubnet)
writeAddresses(w, addrs)
})
router.Methods("POST").Path("/ip/{id}/{ip}/{prefixlen}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
if subnet, ok := parseCIDR(w, vars["ip"]+"/"+vars["prefixlen"]); ok {
if subnet, ok := parseCIDR(w, vars["ip"]+"/"+vars["prefixlen"], true); ok {
alloc.handleHTTPAllocate(dockerCli, w, vars["id"], r.FormValue("check-alive") == "true", subnet)
}
})
ipam/status.go
+3 -3
View File
@@ -26,8 +26,8 @@ type EntryStatus struct {
}
type ClaimStatus struct {
Ident string
Address address.Address
Ident string
CIDR address.CIDR
}
func NewStatus(allocator *Allocator, defaultSubnet address.CIDR) *Status {
@@ -80,7 +80,7 @@ func newClaimStatusSlice(allocator *Allocator) []ClaimStatus {
var slice []ClaimStatus
for _, op := range allocator.pendingClaims {
claim := op.(*claim)
slice = append(slice, ClaimStatus{claim.ident, claim.addr})
slice = append(slice, ClaimStatus{claim.ident, claim.cidr})
}
return slice
}
net/address/address.go
+10 -6
View File
@@ -31,11 +31,15 @@ func (r Range) AsCIDRString() string {
}
prefixLen--
}
return CIDR{Start: r.Start, PrefixLen: prefixLen}.String()
return CIDR{Addr: r.Start, PrefixLen: prefixLen}.String()
}
func MakeCIDR(subnet CIDR, addr Address) CIDR {
return CIDR{Addr: addr, PrefixLen: subnet.PrefixLen}
}
type CIDR struct {
Start Address
Addr Address
PrefixLen int
}
@@ -53,22 +57,22 @@ func ParseCIDR(s string) (Address, CIDR, error) {
return 0, CIDR{}, &net.ParseError{Type: "Non-IPv4 address not supported", Text: s}
} else {
prefixLen, _ := ipnet.Mask.Size()
return FromIP4(ip), CIDR{Start: FromIP4(ipnet.IP), PrefixLen: prefixLen}, nil
return FromIP4(ip), CIDR{Addr: FromIP4(ipnet.IP), PrefixLen: prefixLen}, nil
}
}
func (cidr CIDR) Size() Offset { return 1 << uint(32-cidr.PrefixLen) }
func (cidr CIDR) Range() Range {
return NewRange(cidr.Start, cidr.Size())
return NewRange(cidr.Addr, cidr.Size())
}
func (cidr CIDR) HostRange() Range {
// Respect RFC1122 exclusions of first and last addresses
return NewRange(cidr.Start+1, cidr.Size()-2)
return NewRange(cidr.Addr+1, cidr.Size()-2)
}
func (cidr CIDR) String() string {
return fmt.Sprintf("%s/%d", cidr.Start.String(), cidr.PrefixLen)
return fmt.Sprintf("%s/%d", cidr.Addr.String(), cidr.PrefixLen)
}
// FromIP4 converts an ipv4 address to our integer address type
weave
+3 -3
View File
@@ -1298,7 +1298,7 @@ check_overlap() {
# with_container_addresses.
ipam_reclaim() {
for CIDR in $3 ; do
http_call $HTTP_ADDR PUT /ip/$1/${CIDR%/*}?noErrorOnUnknown=true
http_call $HTTP_ADDR PUT /ip/$1/$CIDR?noErrorOnUnknown=true
done
}
@@ -1354,7 +1354,7 @@ ipam_cidrs() {
if [ "$METHOD" = "POST" ] ; then
# Assignment of a plain IP address; warn if it clashes but carry on
check_overlap $arg || true
when_weave_running http_call $HTTP_ADDR PUT /ip/$CONTAINER_ID/${arg%/*} >&2
when_weave_running http_call $HTTP_ADDR PUT /ip/$CONTAINER_ID/$arg >&2
fi
ALL_CIDRS="$ALL_CIDRS $arg"
fi
@@ -2064,7 +2064,7 @@ EOF
RES=$(docker restart $1)
CONTAINER=$(container_id $1)
for CIDR in $ALL_CIDRS ; do
call_weave PUT /ip/$CONTAINER/${CIDR%/*}
call_weave PUT /ip/$CONTAINER/$CIDR
done
with_container_netns_or_die $1 attach $ALL_CIDRS
when_weave_running with_container_fqdn $1 put_dns_fqdn $ALL_CIDRS