store.go

package libnetwork

import (
	"encoding/json"
	"fmt"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/docker/libkv/store"
	"github.com/docker/libnetwork/config"
	"github.com/docker/libnetwork/datastore"
)

var (
	defaultBoltTimeout      = 3 * time.Second
	defaultLocalStoreConfig = config.DatastoreCfg{
		Embedded: true,
		Client: config.DatastoreClientCfg{
			Provider: "boltdb",
			Address:  defaultPrefix + "/boltdb.db",
			Config: &store.Config{
				Bucket:            "libnetwork",
				ConnectionTimeout: defaultBoltTimeout,
			},
		},
	}
)

func (c *controller) validateGlobalStoreConfig() bool {
	return c.cfg != nil && c.cfg.GlobalStore.Client.Provider != "" && c.cfg.GlobalStore.Client.Address != ""
}

func (c *controller) initGlobalStore() error {
	c.Lock()
	cfg := c.cfg
	c.Unlock()
	if !c.validateGlobalStoreConfig() {
		return fmt.Errorf("globalstore initialization requires a valid configuration")
	}

	store, err := datastore.NewDataStore(&cfg.GlobalStore)
	if err != nil {
		return err
	}
	c.Lock()
	c.globalStore = store
	c.Unlock()

	nws, err := c.getNetworksFromStore(true)
	if err == nil {
		c.processNetworkUpdate(nws, nil)
	} else if err != datastore.ErrKeyNotFound {
		log.Warnf("failed to read networks from globalstore during init : %v", err)
	}
	return c.watchNetworks()
}

func (c *controller) initLocalStore() error {
	c.Lock()
	cfg := c.cfg
	c.Unlock()
	localStore, err := datastore.NewDataStore(c.getLocalStoreConfig(cfg))
	if err != nil {
		return err
	}
	c.Lock()
	c.localStore = localStore
	c.Unlock()

	nws, err := c.getNetworksFromStore(false)
	if err == nil {
		c.processNetworkUpdate(nws, nil)
	} else if err != datastore.ErrKeyNotFound {
		log.Warnf("failed to read networks from localstore during init : %v", err)
	}
	return nil
}

func (c *controller) getNetworksFromStore(global bool) ([]*store.KVPair, error) {
	var cs datastore.DataStore
	c.Lock()
	if global {
		cs = c.globalStore
	} else {
		cs = c.localStore
	}
	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) newEndpointFromStore(key string, ep *endpoint) error {
	ep.Lock()
	n := ep.network
	id := ep.id
	ep.Unlock()

	_, err := n.EndpointByID(id)
	if err != nil {
		if _, ok := err.(ErrNoSuchEndpoint); ok {
			return n.addEndpoint(ep)
		}
	}
	return err
}

func (c *controller) updateToStore(kvObject datastore.KV) error {
	if kvObject.Skip() {
		return nil
	}
	cs := c.getDataStore(kvObject.DataScope())
	if cs == nil {
		log.Debugf("datastore not initialized. kv object %s is not added to the store", datastore.Key(kvObject.Key()...))
		return nil
	}

	return cs.PutObjectAtomic(kvObject)
}

func (c *controller) deleteFromStore(kvObject datastore.KV) error {
	if kvObject.Skip() {
		return nil
	}
	cs := c.getDataStore(kvObject.DataScope())
	if cs == nil {
		log.Debugf("datastore not initialized. kv object %s is not deleted from datastore", datastore.Key(kvObject.Key()...))
		return nil
	}

	if err := cs.DeleteObjectAtomic(kvObject); err != nil {
		return err
	}

	return nil
}

func (c *controller) watchNetworks() error {
	if !c.validateGlobalStoreConfig() {
		return nil
	}

	c.Lock()
	cs := c.globalStore
	c.Unlock()

	networkKey := datastore.Key(datastore.NetworkKeyPrefix)
	if err := ensureKeys(networkKey, cs); err != nil {
		return fmt.Errorf("failed to ensure if the network keys are valid and present in store: %v", err)
	}
	nwPairs, err := cs.KVStore().WatchTree(networkKey, 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 {
					if v.isGlobalScoped() {
						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.globalStore.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.Skip() || !n.ctrlr.validateGlobalStoreConfig() {
		return nil
	}

	n.Lock()
	cs := n.ctrlr.globalStore
	tmp := endpoint{network: n}
	n.stopWatchCh = make(chan struct{})
	stopCh := n.stopWatchCh
	n.Unlock()

	endpointKey := datastore.Key(tmp.KeyPrefix()...)
	if err := ensureKeys(endpointKey, cs); err != nil {
		return fmt.Errorf("failed to ensure if the endpoint keys are valid and present in store: %v", err)
	}
	epPairs, err := cs.KVStore().WatchTree(endpointKey, 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 {
					if v.network.isGlobalScoped() {
						tmpview[k] = v
					}
				}
				n.ctrlr.processEndpointsUpdate(eps, &tmpview)
				// 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(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
		ee.sandboxID = ep.sandboxID
	}
	ee.Unlock()

	return false
}

func ensureKeys(key string, cs datastore.DataStore) error {
	exists, err := cs.KVStore().Exists(key)
	if err != nil {
		return err
	}
	if exists {
		return nil
	}
	return cs.KVStore().Put(key, []byte{}, nil)
}

func (c *controller) getLocalStoreConfig(cfg *config.Config) *config.DatastoreCfg {
	if cfg != nil && cfg.LocalStore.Client.Provider != "" && cfg.LocalStore.Client.Address != "" {
		return &cfg.LocalStore
	}
	return &defaultLocalStoreConfig
}

func (c *controller) getDataStore(dataScope datastore.DataScope) (dataStore datastore.DataStore) {
	c.Lock()
	if dataScope == datastore.GlobalScope {
		dataStore = c.globalStore
	} else if dataScope == datastore.LocalScope {
		dataStore = c.localStore
	}
	c.Unlock()
	return
}

func (c *controller) processEndpointsUpdate(eps []*store.KVPair, prune *endpointTable) {
	for _, epe := range eps {
		var ep endpoint
		err := json.Unmarshal(epe.Value, &ep)
		if err != nil {
			log.Error(err)
			continue
		}
		if prune != nil {
			delete(*prune, ep.id)
		}
		ep.SetIndex(epe.LastIndex)
		if nid, err := ep.networkIDFromKey(epe.Key); err != nil {
			log.Error(err)
			continue
		} else {
			if n, err := c.NetworkByID(nid); err != nil {
				log.Error(err)
				continue
			} else {
				ep.network = n.(*network)
			}
		}
		if c.processEndpointUpdate(&ep) {
			err = c.newEndpointFromStore(epe.Key, &ep)
			if err != nil {
				log.Error(err)
			}
		}
	}
}