diff --git a/hack/vendor.sh b/hack/vendor.sh index d14c292bd..98149b45c 100755 --- a/hack/vendor.sh +++ b/hack/vendor.sh @@ -20,7 +20,7 @@ clone git github.com/tchap/go-patricia v2.1.0 clone git golang.org/x/net 3cffabab72adf04f8e3b01c5baf775361837b5fe https://github.com/golang/net.git #get libnetwork packages -clone git github.com/docker/libnetwork f3c8ebf46b890d4612c5d98e792280d13abdb761 +clone git github.com/docker/libnetwork bf041154d27ed34ed39722328c8f1b0144a56fe2 clone git github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec clone git github.com/hashicorp/go-msgpack 71c2886f5a673a35f909803f38ece5810165097b clone git github.com/hashicorp/memberlist 9a1e242e454d2443df330bdd51a436d5a9058fc4 diff --git a/vendor/src/github.com/docker/libnetwork/controller.go b/vendor/src/github.com/docker/libnetwork/controller.go index be027f498..d8a6d76f8 100644 --- a/vendor/src/github.com/docker/libnetwork/controller.go +++ b/vendor/src/github.com/docker/libnetwork/controller.go @@ -218,7 +218,14 @@ func (c *controller) initDiscovery(watcher discovery.Watcher) error { } c.discovery = hostdiscovery.NewHostDiscovery(watcher) - return c.discovery.Watch(c.hostJoinCallback, c.hostLeaveCallback) + return c.discovery.Watch(c.activeCallback, c.hostJoinCallback, c.hostLeaveCallback) +} + +func (c *controller) activeCallback() { + ds := c.getStore(datastore.GlobalScope) + if ds != nil && !ds.Active() { + ds.RestartWatch() + } } func (c *controller) hostJoinCallback(nodes []net.IP) { diff --git a/vendor/src/github.com/docker/libnetwork/datastore/datastore.go b/vendor/src/github.com/docker/libnetwork/datastore/datastore.go index 70be5c9cc..3919afe0f 100644 --- a/vendor/src/github.com/docker/libnetwork/datastore/datastore.go +++ b/vendor/src/github.com/docker/libnetwork/datastore/datastore.go @@ -34,6 +34,10 @@ type DataStore interface { Watchable() bool // Watch for changes on a KVObject Watch(kvObject KVObject, stopCh <-chan struct{}) (<-chan KVObject, error) + // RestartWatch retriggers stopped Watches + RestartWatch() + // Active returns if the store is active + Active() bool // List returns of a list of KVObjects belonging to the parent // key. The caller must pass a KVObject of the same type as // the objects that need to be listed @@ -53,9 +57,11 @@ var ( ) type datastore struct { - scope string - store store.Store - cache *cache + scope string + store store.Store + cache *cache + watchCh chan struct{} + active bool sync.Mutex } @@ -204,7 +210,7 @@ func newClient(scope string, kv string, addr string, config *store.Config, cache return nil, err } - ds := &datastore{scope: scope, store: store} + ds := &datastore{scope: scope, store: store, active: true, watchCh: make(chan struct{})} if cached { ds.cache = newCache(ds) } @@ -239,6 +245,10 @@ func (ds *datastore) Scope() string { return ds.scope } +func (ds *datastore) Active() bool { + return ds.active +} + func (ds *datastore) Watchable() bool { return ds.scope != LocalScope } @@ -259,6 +269,15 @@ func (ds *datastore) Watch(kvObject KVObject, stopCh <-chan struct{}) (<-chan KV kvoCh := make(chan KVObject) go func() { + retry_watch: + var err error + + // Make sure to get a new instance of watch channel + ds.Lock() + watchCh := ds.watchCh + ds.Unlock() + + loop: for { select { case <-stopCh: @@ -269,12 +288,15 @@ func (ds *datastore) Watch(kvObject KVObject, stopCh <-chan struct{}) (<-chan KV // for the watch can exit resulting in a nil value in // channel. if kvPair == nil { - close(sCh) - return + ds.Lock() + ds.active = false + ds.Unlock() + break loop } + dstO := ctor.New() - if err := dstO.SetValue(kvPair.Value); err != nil { + if err = dstO.SetValue(kvPair.Value); err != nil { log.Printf("Could not unmarshal kvpair value = %s", string(kvPair.Value)) break } @@ -283,11 +305,31 @@ func (ds *datastore) Watch(kvObject KVObject, stopCh <-chan struct{}) (<-chan KV kvoCh <- dstO } } + + // Wait on watch channel for a re-trigger when datastore becomes active + <-watchCh + + kvpCh, err = ds.store.Watch(Key(kvObject.Key()...), sCh) + if err != nil { + log.Printf("Could not watch the key %s in store: %v", Key(kvObject.Key()...), err) + } + + goto retry_watch }() return kvoCh, nil } +func (ds *datastore) RestartWatch() { + ds.Lock() + defer ds.Unlock() + + ds.active = true + watchCh := ds.watchCh + ds.watchCh = make(chan struct{}) + close(watchCh) +} + func (ds *datastore) KVStore() store.Store { return ds.store } diff --git a/vendor/src/github.com/docker/libnetwork/etchosts/etchosts.go b/vendor/src/github.com/docker/libnetwork/etchosts/etchosts.go index 466143bae..92597b71b 100644 --- a/vendor/src/github.com/docker/libnetwork/etchosts/etchosts.go +++ b/vendor/src/github.com/docker/libnetwork/etchosts/etchosts.go @@ -7,6 +7,7 @@ import ( "io/ioutil" "os" "regexp" + "sync" ) // Record Structure for a single host record @@ -21,14 +22,47 @@ func (r Record) WriteTo(w io.Writer) (int64, error) { return int64(n), err } -// Default hosts config records slice -var defaultContent = []Record{ - {Hosts: "localhost", IP: "127.0.0.1"}, - {Hosts: "localhost ip6-localhost ip6-loopback", IP: "::1"}, - {Hosts: "ip6-localnet", IP: "fe00::0"}, - {Hosts: "ip6-mcastprefix", IP: "ff00::0"}, - {Hosts: "ip6-allnodes", IP: "ff02::1"}, - {Hosts: "ip6-allrouters", IP: "ff02::2"}, +var ( + // Default hosts config records slice + defaultContent = []Record{ + {Hosts: "localhost", IP: "127.0.0.1"}, + {Hosts: "localhost ip6-localhost ip6-loopback", IP: "::1"}, + {Hosts: "ip6-localnet", IP: "fe00::0"}, + {Hosts: "ip6-mcastprefix", IP: "ff00::0"}, + {Hosts: "ip6-allnodes", IP: "ff02::1"}, + {Hosts: "ip6-allrouters", IP: "ff02::2"}, + } + + // A cache of path level locks for synchronizing /etc/hosts + // updates on a file level + pathMap = make(map[string]*sync.Mutex) + + // A package level mutex to synchronize the cache itself + pathMutex sync.Mutex +) + +func pathLock(path string) func() { + pathMutex.Lock() + defer pathMutex.Unlock() + + pl, ok := pathMap[path] + if !ok { + pl = &sync.Mutex{} + pathMap[path] = pl + } + + pl.Lock() + return func() { + pl.Unlock() + } +} + +// Drop drops the path string from the path cache +func Drop(path string) { + pathMutex.Lock() + defer pathMutex.Unlock() + + delete(pathMap, path) } // Build function @@ -36,6 +70,8 @@ var defaultContent = []Record{ // IP, hostname, and domainname set main record leave empty for no master record // extraContent is an array of extra host records. func Build(path, IP, hostname, domainname string, extraContent []Record) error { + defer pathLock(path)() + content := bytes.NewBuffer(nil) if IP != "" { //set main record @@ -68,6 +104,8 @@ func Build(path, IP, hostname, domainname string, extraContent []Record) error { // Add adds an arbitrary number of Records to an already existing /etc/hosts file func Add(path string, recs []Record) error { + defer pathLock(path)() + if len(recs) == 0 { return nil } @@ -95,6 +133,8 @@ func Add(path string, recs []Record) error { // Delete deletes an arbitrary number of Records already existing in /etc/hosts file func Delete(path string, recs []Record) error { + defer pathLock(path)() + if len(recs) == 0 { return nil } @@ -118,6 +158,8 @@ func Delete(path string, recs []Record) error { // IP is new IP address // hostname is hostname to search for to replace IP func Update(path, IP, hostname string) error { + defer pathLock(path)() + old, err := ioutil.ReadFile(path) if err != nil { return err diff --git a/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery.go b/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery.go index cb29e4503..3fe2a64a1 100644 --- a/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery.go +++ b/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery.go @@ -34,7 +34,7 @@ func NewHostDiscovery(watcher discovery.Watcher) HostDiscovery { return &hostDiscovery{watcher: watcher, nodes: mapset.NewSet(), stopChan: make(chan struct{})} } -func (h *hostDiscovery) Watch(joinCallback JoinCallback, leaveCallback LeaveCallback) error { +func (h *hostDiscovery) Watch(activeCallback ActiveCallback, joinCallback JoinCallback, leaveCallback LeaveCallback) error { h.Lock() d := h.watcher h.Unlock() @@ -42,15 +42,16 @@ func (h *hostDiscovery) Watch(joinCallback JoinCallback, leaveCallback LeaveCall return types.BadRequestErrorf("invalid discovery watcher") } discoveryCh, errCh := d.Watch(h.stopChan) - go h.monitorDiscovery(discoveryCh, errCh, joinCallback, leaveCallback) + go h.monitorDiscovery(discoveryCh, errCh, activeCallback, joinCallback, leaveCallback) return nil } -func (h *hostDiscovery) monitorDiscovery(ch <-chan discovery.Entries, errCh <-chan error, joinCallback JoinCallback, leaveCallback LeaveCallback) { +func (h *hostDiscovery) monitorDiscovery(ch <-chan discovery.Entries, errCh <-chan error, + activeCallback ActiveCallback, joinCallback JoinCallback, leaveCallback LeaveCallback) { for { select { case entries := <-ch: - h.processCallback(entries, joinCallback, leaveCallback) + h.processCallback(entries, activeCallback, joinCallback, leaveCallback) case err := <-errCh: if err != nil { log.Errorf("discovery error: %v", err) @@ -71,7 +72,8 @@ func (h *hostDiscovery) StopDiscovery() error { return nil } -func (h *hostDiscovery) processCallback(entries discovery.Entries, joinCallback JoinCallback, leaveCallback LeaveCallback) { +func (h *hostDiscovery) processCallback(entries discovery.Entries, + activeCallback ActiveCallback, joinCallback JoinCallback, leaveCallback LeaveCallback) { updated := hosts(entries) h.Lock() existing := h.nodes @@ -79,6 +81,7 @@ func (h *hostDiscovery) processCallback(entries discovery.Entries, joinCallback h.nodes = updated h.Unlock() + activeCallback() if len(added) > 0 { joinCallback(added) } diff --git a/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery_api.go b/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery_api.go index 5be520fca..b9c17250c 100644 --- a/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery_api.go +++ b/vendor/src/github.com/docker/libnetwork/hostdiscovery/hostdiscovery_api.go @@ -5,13 +5,16 @@ import "net" // JoinCallback provides a callback event for new node joining the cluster type JoinCallback func(entries []net.IP) +// ActiveCallback provides a callback event for active discovery event +type ActiveCallback func() + // LeaveCallback provides a callback event for node leaving the cluster type LeaveCallback func(entries []net.IP) // HostDiscovery primary interface type HostDiscovery interface { //Watch Node join and leave cluster events - Watch(joinCallback JoinCallback, leaveCallback LeaveCallback) error + Watch(activeCallback ActiveCallback, joinCallback JoinCallback, leaveCallback LeaveCallback) error // StopDiscovery stops the discovery perocess StopDiscovery() error // Fetch returns a list of host IPs that are currently discovered diff --git a/vendor/src/github.com/docker/libnetwork/ipam/allocator.go b/vendor/src/github.com/docker/libnetwork/ipam/allocator.go index 119031da8..7e4d85697 100644 --- a/vendor/src/github.com/docker/libnetwork/ipam/allocator.go +++ b/vendor/src/github.com/docker/libnetwork/ipam/allocator.go @@ -250,11 +250,6 @@ func (a *Allocator) insertBitMask(key SubnetKey, pool *net.IPNet) error { ones, bits := pool.Mask.Size() numAddresses := uint64(1 << uint(bits-ones)) - if ipVer == v4 { - // Do not let broadcast address be reserved - numAddresses-- - } - // Allow /64 subnet if ipVer == v6 && numAddresses == 0 { numAddresses-- @@ -270,6 +265,11 @@ func (a *Allocator) insertBitMask(key SubnetKey, pool *net.IPNet) error { // Do the same for IPv6 so that bridge ip starts with XXXX...::1 h.Set(0) + // Do not let broadcast address be reserved + if ipVer == v4 { + h.Set(numAddresses - 1) + } + a.Lock() a.addresses[key] = h a.Unlock() diff --git a/vendor/src/github.com/docker/libnetwork/ipams/remote/remote.go b/vendor/src/github.com/docker/libnetwork/ipams/remote/remote.go index f9df525ce..3aefd430a 100644 --- a/vendor/src/github.com/docker/libnetwork/ipams/remote/remote.go +++ b/vendor/src/github.com/docker/libnetwork/ipams/remote/remote.go @@ -78,7 +78,11 @@ func (a *allocator) ReleasePool(poolID string) error { // RequestAddress requests an address from the address pool func (a *allocator) RequestAddress(poolID string, address net.IP, options map[string]string) (*net.IPNet, map[string]string, error) { - var prefAddress string + var ( + prefAddress string + retAddress *net.IPNet + err error + ) if address != nil { prefAddress = address.String() } @@ -87,7 +91,9 @@ func (a *allocator) RequestAddress(poolID string, address net.IP, options map[st if err := a.call("RequestAddress", req, res); err != nil { return nil, nil, err } - retAddress, err := types.ParseCIDR(res.Address) + if res.Address != "" { + retAddress, err = types.ParseCIDR(res.Address) + } return retAddress, res.Data, err } diff --git a/vendor/src/github.com/docker/libnetwork/sandbox.go b/vendor/src/github.com/docker/libnetwork/sandbox.go index 7b30ade61..ff0c66fc9 100644 --- a/vendor/src/github.com/docker/libnetwork/sandbox.go +++ b/vendor/src/github.com/docker/libnetwork/sandbox.go @@ -182,6 +182,10 @@ func (sb *sandbox) Delete() error { } } + // Container is going away. Path cache in etchosts is most + // likely not required any more. Drop it. + etchosts.Drop(sb.config.hostsPath) + if sb.osSbox != nil { sb.osSbox.Destroy() } diff --git a/vendor/src/github.com/docker/libnetwork/store.go b/vendor/src/github.com/docker/libnetwork/store.go index c70caf08d..bacc6170c 100644 --- a/vendor/src/github.com/docker/libnetwork/store.go +++ b/vendor/src/github.com/docker/libnetwork/store.go @@ -308,6 +308,11 @@ func (c *controller) processEndpointCreate(nmap map[string]*netWatch, ep *endpoi c.Lock() nw.localEps[ep.ID()] = ep + + // If we had learned that from the kv store remove it + // from remote ep list now that we know that this is + // indeed a local endpoint + delete(nw.remoteEps, ep.ID()) c.Unlock() return }