package libnetwork import ( "encoding/json" "fmt" log "github.com/Sirupsen/logrus" "github.com/docker/libkv/store" "github.com/docker/libnetwork/datastore" "github.com/docker/libnetwork/types" ) func (c *controller) validateDatastoreConfig() bool { if c.cfg == nil || c.cfg.Datastore.Client.Provider == "" || c.cfg.Datastore.Client.Address == "" { return false } return true } func (c *controller) initDataStore() error { c.Lock() cfg := c.cfg c.Unlock() if !c.validateDatastoreConfig() { return fmt.Errorf("datastore initialization requires a valid configuration") } store, err := datastore.NewDataStore(&cfg.Datastore) if err != nil { return err } c.Lock() c.store = store c.Unlock() nws, err := c.getNetworksFromStore() if err == nil { c.processNetworkUpdate(nws, nil) } else if err != datastore.ErrKeyNotFound { log.Warnf("failed to read networks from datastore during init : %v", err) } return c.watchNetworks() } func (c *controller) getNetworksFromStore() ([]*store.KVPair, error) { c.Lock() cs := c.store c.Unlock() return cs.KVStore().List(datastore.Key(datastore.NetworkKeyPrefix)) } func (c *controller) newNetworkFromStore(n *network) error { n.Lock() n.ctrlr = c n.endpoints = endpointTable{} n.Unlock() return c.addNetwork(n) } func (c *controller) updateNetworkToStore(n *network) error { global, err := n.isGlobalScoped() if err != nil || !global { return err } c.Lock() cs := c.store c.Unlock() if cs == nil { log.Debugf("datastore not initialized. Network %s is not added to the store", n.Name()) return nil } return cs.PutObjectAtomic(n) } func (c *controller) deleteNetworkFromStore(n *network) error { global, err := n.isGlobalScoped() if err != nil || !global { return err } c.Lock() cs := c.store c.Unlock() if cs == nil { log.Debugf("datastore not initialized. Network %s is not deleted from datastore", n.Name()) return nil } if err := cs.DeleteObjectAtomic(n); err != nil { return err } return nil } func (c *controller) getNetworkFromStore(nid types.UUID) (*network, error) { n := network{id: nid} if err := c.store.GetObject(datastore.Key(n.Key()...), &n); err != nil { return nil, err } return &n, nil } func (c *controller) newEndpointFromStore(key string, ep *endpoint) error { ep.Lock() n := ep.network id := ep.id ep.Unlock() _, err := n.EndpointByID(string(id)) if err != nil { if _, ok := err.(ErrNoSuchEndpoint); ok { return n.addEndpoint(ep) } } return err } func (c *controller) updateEndpointToStore(ep *endpoint) error { ep.Lock() n := ep.network name := ep.name ep.Unlock() global, err := n.isGlobalScoped() if err != nil || !global { return err } c.Lock() cs := c.store c.Unlock() if cs == nil { log.Debugf("datastore not initialized. endpoint %s is not added to the store", name) return nil } return cs.PutObjectAtomic(ep) } func (c *controller) getEndpointFromStore(eid types.UUID) (*endpoint, error) { ep := endpoint{id: eid} if err := c.store.GetObject(datastore.Key(ep.Key()...), &ep); err != nil { return nil, err } return &ep, nil } func (c *controller) deleteEndpointFromStore(ep *endpoint) error { ep.Lock() n := ep.network ep.Unlock() global, err := n.isGlobalScoped() if err != nil || !global { return err } c.Lock() cs := c.store c.Unlock() if cs == nil { log.Debugf("datastore not initialized. endpoint %s is not deleted from datastore", ep.Name()) return nil } if err := cs.DeleteObjectAtomic(ep); err != nil { return err } return nil } func (c *controller) watchNetworks() error { if !c.validateDatastoreConfig() { return nil } c.Lock() cs := c.store c.Unlock() nwPairs, err := cs.KVStore().WatchTree(datastore.Key(datastore.NetworkKeyPrefix), nil) if err != nil { return err } go func() { for { select { case nws := <-nwPairs: c.Lock() tmpview := networkTable{} lview := c.networks c.Unlock() for k, v := range lview { global, _ := v.isGlobalScoped() if global { tmpview[k] = v } } c.processNetworkUpdate(nws, &tmpview) // Delete processing for k := range tmpview { c.Lock() existing, ok := c.networks[k] c.Unlock() if !ok { continue } tmp := network{} if err := c.store.GetObject(datastore.Key(existing.Key()...), &tmp); err != datastore.ErrKeyNotFound { continue } if err := existing.deleteNetwork(); err != nil { log.Debugf("Delete failed %s: %s", existing.name, err) } } } } }() return nil } func (n *network) watchEndpoints() error { if !n.ctrlr.validateDatastoreConfig() { return nil } n.Lock() cs := n.ctrlr.store tmp := endpoint{network: n} n.stopWatchCh = make(chan struct{}) stopCh := n.stopWatchCh n.Unlock() epPairs, err := cs.KVStore().WatchTree(datastore.Key(tmp.KeyPrefix()...), stopCh) if err != nil { return err } go func() { for { select { case <-stopCh: return case eps := <-epPairs: n.Lock() tmpview := endpointTable{} lview := n.endpoints n.Unlock() for k, v := range lview { global, _ := v.network.isGlobalScoped() if global { tmpview[k] = v } } for _, epe := range eps { var ep endpoint err := json.Unmarshal(epe.Value, &ep) if err != nil { log.Error(err) continue } delete(tmpview, ep.id) ep.SetIndex(epe.LastIndex) ep.network = n if n.ctrlr.processEndpointUpdate(&ep) { err = n.ctrlr.newEndpointFromStore(epe.Key, &ep) if err != nil { log.Error(err) } } } // Delete processing for k := range tmpview { n.Lock() existing, ok := n.endpoints[k] n.Unlock() if !ok { continue } tmp := endpoint{} if err := cs.GetObject(datastore.Key(existing.Key()...), &tmp); err != datastore.ErrKeyNotFound { continue } if err := existing.deleteEndpoint(); err != nil { log.Debugf("Delete failed %s: %s", existing.name, err) } } } } }() return nil } func (n *network) stopWatch() { n.Lock() if n.stopWatchCh != nil { close(n.stopWatchCh) n.stopWatchCh = nil } n.Unlock() } func (c *controller) processNetworkUpdate(nws []*store.KVPair, prune *networkTable) { for _, kve := range nws { var n network err := json.Unmarshal(kve.Value, &n) if err != nil { log.Error(err) continue } if prune != nil { delete(*prune, n.id) } n.SetIndex(kve.LastIndex) c.Lock() existing, ok := c.networks[n.id] c.Unlock() if ok { existing.Lock() // Skip existing network update if existing.dbIndex != n.Index() { // Can't use SetIndex() since existing is locked. existing.dbIndex = n.Index() existing.dbExists = true existing.endpointCnt = n.endpointCnt } existing.Unlock() continue } if err = c.newNetworkFromStore(&n); err != nil { log.Error(err) } } } func (c *controller) processEndpointUpdate(ep *endpoint) bool { nw := ep.network if nw == nil { return true } nw.Lock() id := nw.id nw.Unlock() c.Lock() n, ok := c.networks[id] c.Unlock() if !ok { return true } existing, _ := n.EndpointByID(string(ep.id)) if existing == nil { return true } ee := existing.(*endpoint) ee.Lock() if ee.dbIndex != ep.Index() { // Can't use SetIndex() because ee is locked. ee.dbIndex = ep.Index() ee.dbExists = true if ee.container != nil && ep.container != nil { // we care only about the container id ee.container.id = ep.container.id } else { // we still care only about the container id, but this is a short-cut to communicate join or leave operation ee.container = ep.container } } ee.Unlock() return false }