diff --git a/ipam/allocator.go b/ipam/allocator.go index 98f36f3d..9dee326a 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -202,10 +202,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, cancelChan <-chan bool) error { +func (alloc *Allocator) Claim(ident string, addr address.Address) error { resultChan := make(chan error) - op := &claim{resultChan: resultChan, ident: ident, addr: addr, - hasBeenCancelled: hasBeenCancelled(cancelChan)} + op := &claim{resultChan: resultChan, ident: ident, addr: addr} alloc.doOperation(op, &alloc.pendingClaims) return <-resultChan } diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index 33bed602..8f6683ce 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -143,7 +143,7 @@ func TestAllocatorClaim(t *testing.T) { alloc.claimRingForTesting() addr1, _ := address.ParseIP(testAddr1) - err := alloc.Claim(container3, addr1, nil) + err := alloc.Claim(container3, addr1) wt.AssertNoErr(t, err) // Check we get this address back if we try an allocate addr3, _ := alloc.Allocate(container3, subnet, nil) diff --git a/ipam/claim.go b/ipam/claim.go index 8d96e6e2..c1dbc529 100644 --- a/ipam/claim.go +++ b/ipam/claim.go @@ -3,28 +3,35 @@ package ipam import ( "fmt" + "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/ipam/address" "github.com/weaveworks/weave/router" ) type claim struct { - resultChan chan<- error - ident string - addr address.Address - hasBeenCancelled func() bool + resultChan chan<- error + ident string + addr address.Address +} + +func (c *claim) sendResult(result error) { + // Make sure we only send a result once, since listener stops listening after that + if c.resultChan != nil { + c.resultChan <- result + close(c.resultChan) + c.resultChan = nil + } + if result != nil { + common.Error.Println("[allocator] " + 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 (c.hasBeenCancelled)() { - c.Cancel() - return true - } - if !alloc.ring.Contains(c.addr) { // Address not within our universe; assume user knows what they are doing alloc.infof("Ignored address %s claimed by %s - not in our universe\n", c.addr, c.ident) - c.resultChan <- nil + c.sendResult(nil) return true } @@ -34,6 +41,7 @@ func (c *claim) Try(alloc *Allocator) bool { owner := alloc.ring.Owner(c.addr) if owner == router.UnknownPeerName { alloc.infof("Ring is empty; will try later.\n", c.addr, owner) + c.sendResult(nil) // don't make the caller wait return false } if owner != alloc.ourName { @@ -41,34 +49,34 @@ func (c *claim) Try(alloc *Allocator) bool { if found { name = " (" + name + ")" } - c.resultChan <- 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.addr.String(), owner, name)) return true } // We are the owner, check we haven't given it to another container existingIdent := alloc.findOwner(c.addr) if existingIdent == c.ident { // same identifier is claiming same address; that's OK - c.resultChan <- nil + c.sendResult(nil) return true } if existingIdent == "" { err := alloc.space.Claim(c.addr) if err != nil { - c.resultChan <- err + c.sendResult(err) return true } alloc.debugln("Claimed", c.addr, "for", c.ident) alloc.addOwned(c.ident, c.addr) - c.resultChan <- nil + c.sendResult(nil) return true } // Addr already owned by container on this machine - c.resultChan <- 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.addr.String(), existingIdent)) return true } func (c *claim) Cancel() { - c.resultChan <- fmt.Errorf("Operation cancelled.") + c.sendResult(fmt.Errorf("Operation cancelled.")) } func (c *claim) String() string { diff --git a/ipam/http.go b/ipam/http.go index 8164d8c8..c81ee2ba 100644 --- a/ipam/http.go +++ b/ipam/http.go @@ -19,14 +19,13 @@ func badRequest(w http.ResponseWriter, err error) { // HandleHTTP wires up ipams HTTP endpoints to the provided mux. func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, dockerCli *docker.Client) { router.Methods("PUT").Path("/ip/{id}/{ip}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - closedChan := w.(http.CloseNotifier).CloseNotify() vars := mux.Vars(r) ident := vars["id"] ipStr := vars["ip"] if ip, err := address.ParseIP(ipStr); err != nil { badRequest(w, err) return - } else if err := alloc.Claim(ident, ip, closedChan); err != nil { + } else if err := alloc.Claim(ident, ip); err != nil { badRequest(w, fmt.Errorf("Unable to claim: %s", err)) return }