store.go

package libnetwork

import (
	"fmt"
	"strings"

	"github.com/docker/docker/libnetwork/datastore"
	"github.com/docker/libkv/store/boltdb"
	"github.com/sirupsen/logrus"
)

func registerKVStores() {
	boltdb.Register()
}

func (c *Controller) initStores() error {
	registerKVStores()

	c.mu.Lock()
	defer c.mu.Unlock()

	if c.cfg == nil {
		return nil
	}
	var err error
	c.store, err = datastore.NewDataStore(c.cfg.Scope)
	if err != nil {
		return err
	}

	c.startWatch()
	return nil
}

func (c *Controller) closeStores() {
	if store := c.store; store != nil {
		store.Close()
	}
}

func (c *Controller) getStore() datastore.DataStore {
	c.mu.Lock()
	defer c.mu.Unlock()

	return c.store
}

func (c *Controller) getNetworkFromStore(nid string) (*network, error) {
	for _, n := range c.getNetworksFromStore() {
		if n.id == nid {
			return n, nil
		}
	}
	return nil, ErrNoSuchNetwork(nid)
}

func (c *Controller) getNetworks() ([]*network, error) {
	var nl []*network

	store := c.getStore()
	if store == nil {
		return nil, nil
	}

	kvol, err := store.List(datastore.Key(datastore.NetworkKeyPrefix),
		&network{ctrlr: c})
	if err != nil && err != datastore.ErrKeyNotFound {
		return nil, fmt.Errorf("failed to get networks: %w", err)
	}

	for _, kvo := range kvol {
		n := kvo.(*network)
		n.ctrlr = c

		ec := &endpointCnt{n: n}
		err = store.GetObject(datastore.Key(ec.Key()...), ec)
		if err != nil && !n.inDelete {
			logrus.Warnf("Could not find endpoint count key %s for network %s while listing: %v", datastore.Key(ec.Key()...), n.Name(), err)
			continue
		}

		n.epCnt = ec
		if n.scope == "" {
			n.scope = store.Scope()
		}
		nl = append(nl, n)
	}

	return nl, nil
}

func (c *Controller) getNetworksFromStore() []*network { // FIXME: unify with c.getNetworks()
	var nl []*network

	store := c.getStore()
	kvol, err := store.List(datastore.Key(datastore.NetworkKeyPrefix), &network{ctrlr: c})
	if err != nil {
		if err != datastore.ErrKeyNotFound {
			logrus.Debugf("failed to get networks from store: %v", err)
		}
		return nil
	}

	kvep, err := store.Map(datastore.Key(epCntKeyPrefix), &endpointCnt{})
	if err != nil && err != datastore.ErrKeyNotFound {
		logrus.Warnf("failed to get endpoint_count map from store: %v", err)
	}

	for _, kvo := range kvol {
		n := kvo.(*network)
		n.mu.Lock()
		n.ctrlr = c
		ec := &endpointCnt{n: n}
		// Trim the leading & trailing "/" to make it consistent across all stores
		if val, ok := kvep[strings.Trim(datastore.Key(ec.Key()...), "/")]; ok {
			ec = val.(*endpointCnt)
			ec.n = n
			n.epCnt = ec
		}
		if n.scope == "" {
			n.scope = store.Scope()
		}
		n.mu.Unlock()
		nl = append(nl, n)
	}

	return nl
}

func (n *network) getEndpointFromStore(eid string) (*Endpoint, error) {
	store := n.ctrlr.getStore()
	ep := &Endpoint{id: eid, network: n}
	err := store.GetObject(datastore.Key(ep.Key()...), ep)
	if err != nil {
		return nil, fmt.Errorf("could not find endpoint %s: %w", eid, err)
	}
	return ep, nil
}

func (n *network) getEndpointsFromStore() ([]*Endpoint, error) {
	var epl []*Endpoint

	tmp := Endpoint{network: n}
	store := n.getController().getStore()
	kvol, err := store.List(datastore.Key(tmp.KeyPrefix()...), &Endpoint{network: n})
	if err != nil {
		if err != datastore.ErrKeyNotFound {
			return nil, fmt.Errorf("failed to get endpoints for network %s scope %s: %w",
				n.Name(), store.Scope(), err)
		}
		return nil, nil
	}

	for _, kvo := range kvol {
		ep := kvo.(*Endpoint)
		epl = append(epl, ep)
	}

	return epl, nil
}

func (c *Controller) updateToStore(kvObject datastore.KVObject) error {
	cs := c.getStore()
	if cs == nil {
		return ErrDataStoreNotInitialized
	}

	if err := cs.PutObjectAtomic(kvObject); err != nil {
		if err == datastore.ErrKeyModified {
			return err
		}
		return fmt.Errorf("failed to update store for object type %T: %v", kvObject, err)
	}

	return nil
}

func (c *Controller) deleteFromStore(kvObject datastore.KVObject) error {
	cs := c.getStore()
	if cs == nil {
		return ErrDataStoreNotInitialized
	}

retry:
	if err := cs.DeleteObjectAtomic(kvObject); err != nil {
		if err == datastore.ErrKeyModified {
			if err := cs.GetObject(datastore.Key(kvObject.Key()...), kvObject); err != nil {
				return fmt.Errorf("could not update the kvobject to latest when trying to delete: %v", err)
			}
			logrus.Warnf("Error (%v) deleting object %v, retrying....", err, kvObject.Key())
			goto retry
		}
		return err
	}

	return nil
}

type netWatch struct {
	localEps  map[string]*Endpoint
	remoteEps map[string]*Endpoint
	stopCh    chan struct{}
}

func (c *Controller) getLocalEps(nw *netWatch) []*Endpoint {
	c.mu.Lock()
	defer c.mu.Unlock()

	var epl []*Endpoint
	for _, ep := range nw.localEps {
		epl = append(epl, ep)
	}

	return epl
}

func (c *Controller) watchSvcRecord(ep *Endpoint) {
	c.watchCh <- ep
}

func (c *Controller) unWatchSvcRecord(ep *Endpoint) {
	c.unWatchCh <- ep
}

func (c *Controller) networkWatchLoop(nw *netWatch, ep *Endpoint, ecCh <-chan datastore.KVObject) {
	for {
		select {
		case <-nw.stopCh:
			return
		case o := <-ecCh:
			ec := o.(*endpointCnt)

			epl, err := ec.n.getEndpointsFromStore()
			if err != nil {
				logrus.WithError(err).Debug("error getting endpoints from store")
				continue
			}

			c.mu.Lock()
			var addEp []*Endpoint

			delEpMap := make(map[string]*Endpoint)
			renameEpMap := make(map[string]bool)
			for k, v := range nw.remoteEps {
				delEpMap[k] = v
			}

			for _, lEp := range epl {
				if _, ok := nw.localEps[lEp.ID()]; ok {
					continue
				}

				if ep, ok := nw.remoteEps[lEp.ID()]; ok {
					// On a container rename EP ID will remain
					// the same but the name will change. service
					// records should reflect the change.
					// Keep old EP entry in the delEpMap and add
					// EP from the store (which has the new name)
					// into the new list
					if lEp.name == ep.name {
						delete(delEpMap, lEp.ID())
						continue
					}
					renameEpMap[lEp.ID()] = true
				}
				nw.remoteEps[lEp.ID()] = lEp
				addEp = append(addEp, lEp)
			}

			// EPs whose name are to be deleted from the svc records
			// should also be removed from nw's remote EP list, except
			// the ones that are getting renamed.
			for _, lEp := range delEpMap {
				if !renameEpMap[lEp.ID()] {
					delete(nw.remoteEps, lEp.ID())
				}
			}
			c.mu.Unlock()

			for _, lEp := range delEpMap {
				ep.getNetwork().updateSvcRecord(lEp, c.getLocalEps(nw), false)
			}
			for _, lEp := range addEp {
				ep.getNetwork().updateSvcRecord(lEp, c.getLocalEps(nw), true)
			}
		}
	}
}

func (c *Controller) processEndpointCreate(nmap map[string]*netWatch, ep *Endpoint) {
	n := ep.getNetwork()
	if !c.isDistributedControl() && n.Scope() == datastore.SwarmScope && n.driverIsMultihost() {
		return
	}

	networkID := n.ID()
	endpointID := ep.ID()

	c.mu.Lock()
	nw, ok := nmap[networkID]
	c.mu.Unlock()

	if ok {
		// Update the svc db for the local endpoint join right away
		n.updateSvcRecord(ep, c.getLocalEps(nw), true)

		c.mu.Lock()
		nw.localEps[endpointID] = 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, endpointID)
		c.mu.Unlock()
		return
	}

	nw = &netWatch{
		localEps:  make(map[string]*Endpoint),
		remoteEps: make(map[string]*Endpoint),
	}

	// Update the svc db for the local endpoint join right away
	// Do this before adding this ep to localEps so that we don't
	// try to update this ep's container's svc records
	n.updateSvcRecord(ep, c.getLocalEps(nw), true)

	c.mu.Lock()
	nw.localEps[endpointID] = ep
	nmap[networkID] = nw
	nw.stopCh = make(chan struct{})
	c.mu.Unlock()

	store := c.getStore()
	if store == nil {
		return
	}

	if !store.Watchable() {
		return
	}

	ch, err := store.Watch(n.getEpCnt(), nw.stopCh)
	if err != nil {
		logrus.Warnf("Error creating watch for network: %v", err)
		return
	}

	go c.networkWatchLoop(nw, ep, ch)
}

func (c *Controller) processEndpointDelete(nmap map[string]*netWatch, ep *Endpoint) {
	n := ep.getNetwork()
	if !c.isDistributedControl() && n.Scope() == datastore.SwarmScope && n.driverIsMultihost() {
		return
	}

	networkID := n.ID()
	endpointID := ep.ID()

	c.mu.Lock()
	nw, ok := nmap[networkID]

	if ok {
		delete(nw.localEps, endpointID)
		c.mu.Unlock()

		// Update the svc db about local endpoint leave right away
		// Do this after we remove this ep from localEps so that we
		// don't try to remove this svc record from this ep's container.
		n.updateSvcRecord(ep, c.getLocalEps(nw), false)

		c.mu.Lock()
		if len(nw.localEps) == 0 {
			close(nw.stopCh)

			// This is the last container going away for the network. Destroy
			// this network's svc db entry
			delete(c.svcRecords, networkID)

			delete(nmap, networkID)
		}
	}
	c.mu.Unlock()
}

func (c *Controller) watchLoop() {
	for {
		select {
		case ep := <-c.watchCh:
			c.processEndpointCreate(c.nmap, ep)
		case ep := <-c.unWatchCh:
			c.processEndpointDelete(c.nmap, ep)
		}
	}
}

func (c *Controller) startWatch() {
	if c.watchCh != nil {
		return
	}
	c.watchCh = make(chan *Endpoint)
	c.unWatchCh = make(chan *Endpoint)
	c.nmap = make(map[string]*netWatch)

	go c.watchLoop()
}

func (c *Controller) networkCleanup() {
	for _, n := range c.getNetworksFromStore() {
		if n.inDelete {
			logrus.Infof("Removing stale network %s (%s)", n.Name(), n.ID())
			if err := n.delete(true, true); err != nil {
				logrus.Debugf("Error while removing stale network: %v", err)
			}
		}
	}
}

var populateSpecial NetworkWalker = func(nw Network) bool {
	if n := nw.(*network); n.hasSpecialDriver() && !n.ConfigOnly() {
		if err := n.getController().addNetwork(n); err != nil {
			logrus.Warnf("Failed to populate network %q with driver %q", nw.Name(), nw.Type())
		}
	}
	return false
}