#3248

pending completion

Build # #3248

Build Type

push

circleci

Committed by

aboch

Commit Message

Add missing locks in agent and service code

Run Details

64 of 64 new or added lines in 4 files covered. (100.0%)

11441 of 25981 relevant lines covered (44.04%)

34154.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

7.74

/agent.go

package libnetwork

//go:generate protoc -I.:Godeps/_workspace/src/github.com/gogo/protobuf  --gogo_out=import_path=github.com/docker/libnetwork,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. agent.proto

import (
        "fmt"
        "net"
        "os"
        "sort"

        "github.com/Sirupsen/logrus"
        "github.com/docker/docker/pkg/stringid"
        "github.com/docker/go-events"
        "github.com/docker/libnetwork/datastore"
        "github.com/docker/libnetwork/discoverapi"
        "github.com/docker/libnetwork/driverapi"
        "github.com/docker/libnetwork/networkdb"
        "github.com/docker/libnetwork/types"
        "github.com/gogo/protobuf/proto"
)

const (
        subsysGossip = "networking:gossip"
        subsysIPSec  = "networking:ipsec"
        keyringSize  = 3
)

// ByTime implements sort.Interface for []*types.EncryptionKey based on
// the LamportTime field.
type ByTime []*types.EncryptionKey

func (b ByTime) Len() int           { return len(b) }
func (b ByTime) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
func (b ByTime) Less(i, j int) bool { return b[i].LamportTime < b[j].LamportTime }

type agent struct {
        networkDB         *networkdb.NetworkDB
        bindAddr          string
        advertiseAddr     string
        epTblCancel       func()
        driverCancelFuncs map[string][]func()
}

func getBindAddr(ifaceName string) (string, error) {
        iface, err := net.InterfaceByName(ifaceName)
        if err != nil {
                return "", fmt.Errorf("failed to find interface %s: %v", ifaceName, err)
        }

        addrs, err := iface.Addrs()
        if err != nil {
                return "", fmt.Errorf("failed to get interface addresses: %v", err)
        }

        for _, a := range addrs {
                addr, ok := a.(*net.IPNet)
                if !ok {
                        continue
                }
                addrIP := addr.IP

                if addrIP.IsLinkLocalUnicast() {
                        continue
                }

                return addrIP.String(), nil
        }

        return "", fmt.Errorf("failed to get bind address")
}

func resolveAddr(addrOrInterface string) (string, error) {
        // Try and see if this is a valid IP address
        if net.ParseIP(addrOrInterface) != nil {
                return addrOrInterface, nil
        }

        addr, err := net.ResolveIPAddr("ip", addrOrInterface)
        if err != nil {
                // If not a valid IP address, it should be a valid interface
                return getBindAddr(addrOrInterface)
        }
        return addr.String(), nil
}

func (c *controller) handleKeyChange(keys []*types.EncryptionKey) error {
        drvEnc := discoverapi.DriverEncryptionUpdate{}

        a := c.getAgent()
        if a == nil {
                logrus.Debug("Skipping key change as agent is nil")
                return nil
        }

        // Find the deleted key. If the deleted key was the primary key,
        // a new primary key should be set before removing if from keyring.
        c.Lock()
        deleted := []byte{}
        j := len(c.keys)
        for i := 0; i < j; {
                same := false
                for _, key := range keys {
                        if same = key.LamportTime == c.keys[i].LamportTime; same {
                                break
                        }
                }
                if !same {
                        cKey := c.keys[i]
                        if cKey.Subsystem == subsysGossip {
                                deleted = cKey.Key
                        }

                        if cKey.Subsystem == subsysIPSec {
                                drvEnc.Prune = cKey.Key
                                drvEnc.PruneTag = cKey.LamportTime
                        }
                        c.keys[i], c.keys[j-1] = c.keys[j-1], c.keys[i]
                        c.keys[j-1] = nil
                        j--
                }
                i++
        }
        c.keys = c.keys[:j]

        // Find the new key and add it to the key ring
        for _, key := range keys {
                same := false
                for _, cKey := range c.keys {
                        if same = cKey.LamportTime == key.LamportTime; same {
                                break
                        }
                }
                if !same {
                        c.keys = append(c.keys, key)
                        if key.Subsystem == subsysGossip {
                                a.networkDB.SetKey(key.Key)
                        }

                        if key.Subsystem == subsysIPSec {
                                drvEnc.Key = key.Key
                                drvEnc.Tag = key.LamportTime
                        }
                }
        }
        c.Unlock()

        key, tag, err := c.getPrimaryKeyTag(subsysGossip)
        if err != nil {
                return err
        }
        a.networkDB.SetPrimaryKey(key)

        key, tag, err = c.getPrimaryKeyTag(subsysIPSec)
        if err != nil {
                return err
        }
        drvEnc.Primary = key
        drvEnc.PrimaryTag = tag

        if len(deleted) > 0 {
                a.networkDB.RemoveKey(deleted)
        }

        c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {
                err := driver.DiscoverNew(discoverapi.EncryptionKeysUpdate, drvEnc)
                if err != nil {
                        logrus.Warnf("Failed to update datapath keys in driver %s: %v", name, err)
                }
                return false
        })

        return nil
}

func (c *controller) agentSetup() error {
        c.Lock()
        clusterProvider := c.cfg.Daemon.ClusterProvider
        agent := c.agent
        c.Unlock()
        bindAddr := clusterProvider.GetLocalAddress()
        advAddr := clusterProvider.GetAdvertiseAddress()
        remote := clusterProvider.GetRemoteAddress()
        remoteAddr, _, _ := net.SplitHostPort(remote)
        listen := clusterProvider.GetListenAddress()
        listenAddr, _, _ := net.SplitHostPort(listen)

        logrus.Infof("Initializing Libnetwork Agent Listen-Addr=%s Local-addr=%s Adv-addr=%s Remote-addr =%s", listenAddr, bindAddr, advAddr, remoteAddr)
        if advAddr != "" && agent == nil {
                if err := c.agentInit(listenAddr, bindAddr, advAddr); err != nil {
                        logrus.Errorf("Error in agentInit : %v", err)
                } else {
                        c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {
                                if capability.DataScope == datastore.GlobalScope {
                                        c.agentDriverNotify(driver)
                                }
                                return false
                        })
                }
        }

        if remoteAddr != "" {
                if err := c.agentJoin(remoteAddr); err != nil {
                        logrus.Errorf("Error in joining gossip cluster : %v(join will be retried in background)", err)
                }
        }

        c.Lock()
        if c.agent != nil && c.agentInitDone != nil {
                close(c.agentInitDone)
                c.agentInitDone = nil
        }
        c.Unlock()

        return nil
}

// For a given subsystem getKeys sorts the keys by lamport time and returns
// slice of keys and lamport time which can used as a unique tag for the keys
func (c *controller) getKeys(subsys string) ([][]byte, []uint64) {
        c.Lock()
        defer c.Unlock()

        sort.Sort(ByTime(c.keys))

        keys := [][]byte{}
        tags := []uint64{}
        for _, key := range c.keys {
                if key.Subsystem == subsys {
                        keys = append(keys, key.Key)
                        tags = append(tags, key.LamportTime)
                }
        }

        keys[0], keys[1] = keys[1], keys[0]
        tags[0], tags[1] = tags[1], tags[0]
        return keys, tags
}

// getPrimaryKeyTag returns the primary key for a given subsystem from the
// list of sorted key and the associated tag
func (c *controller) getPrimaryKeyTag(subsys string) ([]byte, uint64, error) {
        c.Lock()
        defer c.Unlock()
        sort.Sort(ByTime(c.keys))
        keys := []*types.EncryptionKey{}
        for _, key := range c.keys {
                if key.Subsystem == subsys {
                        keys = append(keys, key)
                }
        }
        return keys[1].Key, keys[1].LamportTime, nil
}

func (c *controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr string) error {
        if !c.isAgent() {
                return nil
        }

        bindAddr, err := resolveAddr(bindAddrOrInterface)
        if err != nil {
                return err
        }

        keys, tags := c.getKeys(subsysGossip)
        hostname, _ := os.Hostname()
        nodeName := hostname + "-" + stringid.TruncateID(stringid.GenerateRandomID())
        logrus.Info("Gossip cluster hostname ", nodeName)

        nDB, err := networkdb.New(&networkdb.Config{
                BindAddr:      listenAddr,
                AdvertiseAddr: advertiseAddr,
                NodeName:      nodeName,
                Keys:          keys,
        })

        if err != nil {
                return err
        }

        ch, cancel := nDB.Watch("endpoint_table", "", "")

        c.Lock()
        c.agent = &agent{
                networkDB:         nDB,
                bindAddr:          bindAddr,
                advertiseAddr:     advertiseAddr,
                epTblCancel:       cancel,
                driverCancelFuncs: make(map[string][]func()),
        }
        c.Unlock()

        go c.handleTableEvents(ch, c.handleEpTableEvent)

        drvEnc := discoverapi.DriverEncryptionConfig{}
        keys, tags = c.getKeys(subsysIPSec)
        drvEnc.Keys = keys
        drvEnc.Tags = tags

        c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {
                err := driver.DiscoverNew(discoverapi.EncryptionKeysConfig, drvEnc)
                if err != nil {
                        logrus.Warnf("Failed to set datapath keys in driver %s: %v", name, err)
                }
                return false
        })

        c.WalkNetworks(joinCluster)

        return nil
}

func (c *controller) agentJoin(remote string) error {
        agent := c.getAgent()
        if agent == nil {
                return nil
        }
        return agent.networkDB.Join([]string{remote})
}

func (c *controller) agentDriverNotify(d driverapi.Driver) {
        agent := c.getAgent()
        if agent == nil {
                return
        }

        d.DiscoverNew(discoverapi.NodeDiscovery, discoverapi.NodeDiscoveryData{
                Address:     agent.advertiseAddr,
                BindAddress: agent.bindAddr,
                Self:        true,
        })

        drvEnc := discoverapi.DriverEncryptionConfig{}
        keys, tags := c.getKeys(subsysIPSec)
        drvEnc.Keys = keys
        drvEnc.Tags = tags

        c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {
                err := driver.DiscoverNew(discoverapi.EncryptionKeysConfig, drvEnc)
                if err != nil {
                        logrus.Warnf("Failed to set datapath keys in driver %s: %v", name, err)
                }
                return false
        })

}

func (c *controller) agentClose() {
        // Acquire current agent instance and reset its pointer
        // then run closing functions
        c.Lock()
        agent := c.agent
        c.agent = nil
        c.Unlock()

        if agent == nil {
                return
        }

        for _, cancelFuncs := range agent.driverCancelFuncs {
                for _, cancel := range cancelFuncs {
                        cancel()
                }
        }

        agent.epTblCancel()

        agent.networkDB.Close()
}

func (n *network) isClusterEligible() bool {
        if n.driverScope() != datastore.GlobalScope {
                return false
        }
        return n.getController().getAgent() != nil
}

func (n *network) joinCluster() error {
        if !n.isClusterEligible() {
                return nil
        }

        agent := n.getController().getAgent()
        if agent == nil {
                return nil
        }

        return agent.networkDB.JoinNetwork(n.ID())
}

func (n *network) leaveCluster() error {
        if !n.isClusterEligible() {
                return nil
        }

        agent := n.getController().getAgent()
        if agent == nil {
                return nil
        }

        return agent.networkDB.LeaveNetwork(n.ID())
}

func (ep *endpoint) addDriverInfoToCluster() error {
        n := ep.getNetwork()
        if !n.isClusterEligible() {
                return nil
        }
        if ep.joinInfo == nil {
                return nil
        }

        agent := n.getController().getAgent()
        if agent == nil {
                return nil
        }

        for _, te := range ep.joinInfo.driverTableEntries {
                if err := agent.networkDB.CreateEntry(te.tableName, n.ID(), te.key, te.value); err != nil {
                        return err
                }
        }
        return nil
}

func (ep *endpoint) deleteDriverInfoFromCluster() error {
        n := ep.getNetwork()
        if !n.isClusterEligible() {
                return nil
        }
        if ep.joinInfo == nil {
                return nil
        }

        agent := n.getController().getAgent()
        if agent == nil {
                return nil
        }

        for _, te := range ep.joinInfo.driverTableEntries {
                if err := agent.networkDB.DeleteEntry(te.tableName, n.ID(), te.key); err != nil {
                        return err
                }
        }
        return nil
}

func (ep *endpoint) addServiceInfoToCluster() error {
        n := ep.getNetwork()
        if !n.isClusterEligible() {
                return nil
        }

        c := n.getController()
        agent := c.getAgent()
        if !ep.isAnonymous() && ep.Iface().Address() != nil {
                var ingressPorts []*PortConfig
                if ep.svcID != "" {
                        // Gossip ingress ports only in ingress network.
                        if n.ingress {
                                ingressPorts = ep.ingressPorts
                        }

                        if err := c.addServiceBinding(ep.svcName, ep.svcID, n.ID(), ep.ID(), ep.virtualIP, ingressPorts, ep.svcAliases, ep.Iface().Address().IP); err != nil {
                                return err
                        }
                }

                buf, err := proto.Marshal(&EndpointRecord{
                        Name:         ep.Name(),
                        ServiceName:  ep.svcName,
                        ServiceID:    ep.svcID,
                        VirtualIP:    ep.virtualIP.String(),
                        IngressPorts: ingressPorts,
                        Aliases:      ep.svcAliases,
                        TaskAliases:  ep.myAliases,
                        EndpointIP:   ep.Iface().Address().IP.String(),
                })

                if err != nil {
                        return err
                }

                if agent != nil {
                        if err := agent.networkDB.CreateEntry("endpoint_table", n.ID(), ep.ID(), buf); err != nil {
                                return err
                        }
                }
        }

        return nil
}

func (ep *endpoint) deleteServiceInfoFromCluster() error {
        n := ep.getNetwork()
        if !n.isClusterEligible() {
                return nil
        }

        c := n.getController()
        agent := c.getAgent()

        if !ep.isAnonymous() {
                if ep.svcID != "" && ep.Iface().Address() != nil {
                        var ingressPorts []*PortConfig
                        if n.ingress {
                                ingressPorts = ep.ingressPorts
                        }

                        if err := c.rmServiceBinding(ep.svcName, ep.svcID, n.ID(), ep.ID(), ep.virtualIP, ingressPorts, ep.svcAliases, ep.Iface().Address().IP); err != nil {
                                return err
                        }
                }
                if agent != nil {
                        if err := agent.networkDB.DeleteEntry("endpoint_table", n.ID(), ep.ID()); err != nil {
                                return err
                        }
                }
        }
        return nil
}

func (n *network) addDriverWatches() {
        if !n.isClusterEligible() {
                return
        }

        c := n.getController()
        agent := c.getAgent()
        if agent == nil {
                return
        }
        for _, tableName := range n.driverTables {
                ch, cancel := agent.networkDB.Watch(tableName, n.ID(), "")
                agent.driverCancelFuncs[n.ID()] = append(agent.driverCancelFuncs[n.ID()], cancel)

                go c.handleTableEvents(ch, n.handleDriverTableEvent)
                d, err := n.driver(false)
                if err != nil {
                        logrus.Errorf("Could not resolve driver %s while walking driver tabl: %v", n.networkType, err)
                        return
                }

                agent.networkDB.WalkTable(tableName, func(nid, key string, value []byte) bool {
                        if nid == n.ID() {
                                d.EventNotify(driverapi.Create, nid, tableName, key, value)
                        }

                        return false
                })
        }
}

func (n *network) cancelDriverWatches() {
        if !n.isClusterEligible() {
                return
        }

        c := n.getController()
        c.Lock()
        cancelFuncs := c.agent.driverCancelFuncs[n.ID()]
        delete(c.agent.driverCancelFuncs, n.ID())
        c.Unlock()

        for _, cancel := range cancelFuncs {
                cancel()
        }
}

func (c *controller) handleTableEvents(ch chan events.Event, fn func(events.Event)) {
        for {
                select {
                case ev, ok := <-ch:
                        if !ok {
                                return
                        }

                        fn(ev)
                }
        }
}

func (n *network) handleDriverTableEvent(ev events.Event) {
        d, err := n.driver(false)
        if err != nil {
                logrus.Errorf("Could not resolve driver %s while handling driver table event: %v", n.networkType, err)
                return
        }

        var (
                etype driverapi.EventType
                tname string
                key   string
                value []byte
        )

        switch event := ev.(type) {
        case networkdb.CreateEvent:
                tname = event.Table
                key = event.Key
                value = event.Value
                etype = driverapi.Create
        case networkdb.DeleteEvent:
                tname = event.Table
                key = event.Key
                value = event.Value
                etype = driverapi.Delete
        case networkdb.UpdateEvent:
                tname = event.Table
                key = event.Key
                value = event.Value
                etype = driverapi.Delete
        }

        d.EventNotify(etype, n.ID(), tname, key, value)
}

func (c *controller) handleEpTableEvent(ev events.Event) {
        var (
                nid   string
                eid   string
                value []byte
                isAdd bool
                epRec EndpointRecord
        )

        switch event := ev.(type) {
        case networkdb.CreateEvent:
                nid = event.NetworkID
                eid = event.Key
                value = event.Value
                isAdd = true
        case networkdb.DeleteEvent:
                nid = event.NetworkID
                eid = event.Key
                value = event.Value
        case networkdb.UpdateEvent:
                logrus.Errorf("Unexpected update service table event = %#v", event)
        }

        nw, err := c.NetworkByID(nid)
        if err != nil {
                logrus.Errorf("Could not find network %s while handling service table event: %v", nid, err)
                return
        }
        n := nw.(*network)

        err = proto.Unmarshal(value, &epRec)
        if err != nil {
                logrus.Errorf("Failed to unmarshal service table value: %v", err)
                return
        }

        name := epRec.Name
        svcName := epRec.ServiceName
        svcID := epRec.ServiceID
        vip := net.ParseIP(epRec.VirtualIP)
        ip := net.ParseIP(epRec.EndpointIP)
        ingressPorts := epRec.IngressPorts
        aliases := epRec.Aliases
        taskaliases := epRec.TaskAliases

        if name == "" || ip == nil {
                logrus.Errorf("Invalid endpoint name/ip received while handling service table event %s", value)
                return
        }

        if isAdd {
                if svcID != "" {
                        if err := c.addServiceBinding(svcName, svcID, nid, eid, vip, ingressPorts, aliases, ip); err != nil {
                                logrus.Errorf("Failed adding service binding for value %s: %v", value, err)
                                return
                        }
                }

                n.addSvcRecords(name, ip, nil, true)
                for _, alias := range taskaliases {
                        n.addSvcRecords(alias, ip, nil, true)
                }
        } else {
                if svcID != "" {
                        if err := c.rmServiceBinding(svcName, svcID, nid, eid, vip, ingressPorts, aliases, ip); err != nil {
                                logrus.Errorf("Failed adding service binding for value %s: %v", value, err)
                                return
                        }
                }

                n.deleteSvcRecords(name, ip, nil, true)
                for _, alias := range taskaliases {
                        n.deleteSvcRecords(alias, ip, nil, true)
                }
        }
}

1	package libnetwork
2
3	//go:generate protoc -I.:Godeps/_workspace/src/github.com/gogo/protobuf --gogo_out=import_path=github.com/docker/libnetwork,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto:. agent.proto
4
5	import (
6	"fmt"
7	"net"
8	"os"
9	"sort"
10
11	"github.com/Sirupsen/logrus"
12	"github.com/docker/docker/pkg/stringid"
13	"github.com/docker/go-events"
14	"github.com/docker/libnetwork/datastore"
15	"github.com/docker/libnetwork/discoverapi"
16	"github.com/docker/libnetwork/driverapi"
17	"github.com/docker/libnetwork/networkdb"
18	"github.com/docker/libnetwork/types"
19	"github.com/gogo/protobuf/proto"
20	)
21
22	const (
23	subsysGossip = "networking:gossip"
24	subsysIPSec = "networking:ipsec"
25	keyringSize = 3
26	)
27
28	// ByTime implements sort.Interface for []*types.EncryptionKey based on
29	// the LamportTime field.
30	type ByTime []*types.EncryptionKey
31
32	func (b ByTime) Len() int { return len(b) }	×
33	func (b ByTime) Swap(i, j int) { b[i], b[j] = b[j], b[i] }	×
34	func (b ByTime) Less(i, j int) bool { return b[i].LamportTime < b[j].LamportTime }	×
35
36	type agent struct {
37	networkDB *networkdb.NetworkDB
38	bindAddr string
39	advertiseAddr string
40	epTblCancel func()
41	driverCancelFuncs map[string][]func()
42	}
43
44	func getBindAddr(ifaceName string) (string, error) {	×
45	iface, err := net.InterfaceByName(ifaceName)	×
46	if err != nil {	×
47	return "", fmt.Errorf("failed to find interface %s: %v", ifaceName, err)	×
48	}	×
49
50	addrs, err := iface.Addrs()	×
51	if err != nil {	×
52	return "", fmt.Errorf("failed to get interface addresses: %v", err)	×
53	}	×
54
55	for _, a := range addrs {	×
56	addr, ok := a.(*net.IPNet)	×
57	if !ok {	×
58	continue	×
59	}
60	addrIP := addr.IP	×
61		×
62	if addrIP.IsLinkLocalUnicast() {	×
63	continue	×
64	}
65
66	return addrIP.String(), nil	×
67	}
68
69	return "", fmt.Errorf("failed to get bind address")	×
70	}
71
72	func resolveAddr(addrOrInterface string) (string, error) {	×
73	// Try and see if this is a valid IP address	×
74	if net.ParseIP(addrOrInterface) != nil {	×
75	return addrOrInterface, nil	×
76	}	×
77
78	addr, err := net.ResolveIPAddr("ip", addrOrInterface)	×
79	if err != nil {	×
80	// If not a valid IP address, it should be a valid interface	×
81	return getBindAddr(addrOrInterface)	×
82	}	×
83	return addr.String(), nil	×
84	}
85
86	func (c controller) handleKeyChange(keys []types.EncryptionKey) error {	×
87	drvEnc := discoverapi.DriverEncryptionUpdate{}	×
88		×
89	a := c.getAgent()	×
90	if a == nil {	×
91	logrus.Debug("Skipping key change as agent is nil")	×
92	return nil	×
93	}	×
94
95	// Find the deleted key. If the deleted key was the primary key,
96	// a new primary key should be set before removing if from keyring.
97	c.Lock()	×
98	deleted := []byte{}	×
99	j := len(c.keys)	×
100	for i := 0; i < j; {	×
101	same := false	×
102	for _, key := range keys {	×
103	if same = key.LamportTime == c.keys[i].LamportTime; same {	×
104	break	×
105	}
106	}
107	if !same {	×
108	cKey := c.keys[i]	×
109	if cKey.Subsystem == subsysGossip {	×
110	deleted = cKey.Key	×
111	}	×
112
113	if cKey.Subsystem == subsysIPSec {	×
114	drvEnc.Prune = cKey.Key	×
115	drvEnc.PruneTag = cKey.LamportTime	×
116	}	×
117	c.keys[i], c.keys[j-1] = c.keys[j-1], c.keys[i]	×
118	c.keys[j-1] = nil	×
119	j--	×
120	}
121	i++	×
122	}
123	c.keys = c.keys[:j]	×
124		×
125	// Find the new key and add it to the key ring	×
126	for _, key := range keys {	×
127	same := false	×
128	for _, cKey := range c.keys {	×
129	if same = cKey.LamportTime == key.LamportTime; same {	×
130	break	×
131	}
132	}
133	if !same {	×
134	c.keys = append(c.keys, key)	×
135	if key.Subsystem == subsysGossip {	×
136	a.networkDB.SetKey(key.Key)	×
137	}	×
138
139	if key.Subsystem == subsysIPSec {	×
140	drvEnc.Key = key.Key	×
141	drvEnc.Tag = key.LamportTime	×
142	}	×
143	}
144	}
145	c.Unlock()	×
146		×
147	key, tag, err := c.getPrimaryKeyTag(subsysGossip)	×
148	if err != nil {	×
149	return err	×
150	}	×
151	a.networkDB.SetPrimaryKey(key)	×
152		×
153	key, tag, err = c.getPrimaryKeyTag(subsysIPSec)	×
154	if err != nil {	×
155	return err	×
156	}	×
157	drvEnc.Primary = key	×
158	drvEnc.PrimaryTag = tag	×
159		×
160	if len(deleted) > 0 {	×
161	a.networkDB.RemoveKey(deleted)	×
162	}	×
163
164	c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {	×
165	err := driver.DiscoverNew(discoverapi.EncryptionKeysUpdate, drvEnc)	×
166	if err != nil {	×
167	logrus.Warnf("Failed to update datapath keys in driver %s: %v", name, err)	×
168	}	×
169	return false	×
170	})
171
172	return nil	×
173	}
174
175	func (c *controller) agentSetup() error {	×
176	c.Lock()	×
177	clusterProvider := c.cfg.Daemon.ClusterProvider	×
178	agent := c.agent	×
179	c.Unlock()	×
180	bindAddr := clusterProvider.GetLocalAddress()	×
181	advAddr := clusterProvider.GetAdvertiseAddress()	×
182	remote := clusterProvider.GetRemoteAddress()	×
183	remoteAddr, _, _ := net.SplitHostPort(remote)	×
184	listen := clusterProvider.GetListenAddress()	×
185	listenAddr, _, _ := net.SplitHostPort(listen)	×
186		×
187	logrus.Infof("Initializing Libnetwork Agent Listen-Addr=%s Local-addr=%s Adv-addr=%s Remote-addr =%s", listenAddr, bindAddr, advAddr, remoteAddr)	×
188	if advAddr != "" && agent == nil {	×
189	if err := c.agentInit(listenAddr, bindAddr, advAddr); err != nil {	×
190	logrus.Errorf("Error in agentInit : %v", err)	×
191	} else {	×
192	c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {	×
193	if capability.DataScope == datastore.GlobalScope {	×
194	c.agentDriverNotify(driver)	×
195	}	×
196	return false	×
197	})
198	}
199	}
200
201	if remoteAddr != "" {	×
202	if err := c.agentJoin(remoteAddr); err != nil {	×
203	logrus.Errorf("Error in joining gossip cluster : %v(join will be retried in background)", err)	×
204	}	×
205	}
206
207	c.Lock()	×
208	if c.agent != nil && c.agentInitDone != nil {	×
209	close(c.agentInitDone)	×
210	c.agentInitDone = nil	×
211	}	×
212	c.Unlock()	×
213		×
214	return nil	×
215	}
216
217	// For a given subsystem getKeys sorts the keys by lamport time and returns
218	// slice of keys and lamport time which can used as a unique tag for the keys
219	func (c *controller) getKeys(subsys string) ([][]byte, []uint64) {	×
220	c.Lock()	×
221	defer c.Unlock()	×
222		×
223	sort.Sort(ByTime(c.keys))	×
224		×
225	keys := [][]byte{}	×
226	tags := []uint64{}	×
227	for _, key := range c.keys {	×
228	if key.Subsystem == subsys {	×
229	keys = append(keys, key.Key)	×
230	tags = append(tags, key.LamportTime)	×
231	}	×
232	}
233
234	keys[0], keys[1] = keys[1], keys[0]	×
235	tags[0], tags[1] = tags[1], tags[0]	×
236	return keys, tags	×
237	}
238
239	// getPrimaryKeyTag returns the primary key for a given subsystem from the
240	// list of sorted key and the associated tag
241	func (c *controller) getPrimaryKeyTag(subsys string) ([]byte, uint64, error) {	×
242	c.Lock()	×
243	defer c.Unlock()	×
244	sort.Sort(ByTime(c.keys))	×
245	keys := []*types.EncryptionKey{}	×
246	for _, key := range c.keys {	×
247	if key.Subsystem == subsys {	×
248	keys = append(keys, key)	×
249	}	×
250	}
251	return keys[1].Key, keys[1].LamportTime, nil	×
252	}
253
254	func (c *controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr string) error {	×
255	if !c.isAgent() {	×
256	return nil	×
257	}	×
258
259	bindAddr, err := resolveAddr(bindAddrOrInterface)	×
260	if err != nil {	×
261	return err	×
262	}	×
263
264	keys, tags := c.getKeys(subsysGossip)	×
265	hostname, _ := os.Hostname()	×
266	nodeName := hostname + "-" + stringid.TruncateID(stringid.GenerateRandomID())	×
267	logrus.Info("Gossip cluster hostname ", nodeName)	×
268		×
269	nDB, err := networkdb.New(&networkdb.Config{	×
270	BindAddr: listenAddr,	×
271	AdvertiseAddr: advertiseAddr,	×
272	NodeName: nodeName,	×
273	Keys: keys,	×
274	})	×
275		×
276	if err != nil {	×
277	return err	×
278	}	×
279
280	ch, cancel := nDB.Watch("endpoint_table", "", "")	×
281		×
282	c.Lock()	×
283	c.agent = &agent{	×
284	networkDB: nDB,	×
285	bindAddr: bindAddr,	×
286	advertiseAddr: advertiseAddr,	×
287	epTblCancel: cancel,	×
288	driverCancelFuncs: make(map[string][]func()),	×
289	}	×
290	c.Unlock()	×
291		×
292	go c.handleTableEvents(ch, c.handleEpTableEvent)	×
293		×
294	drvEnc := discoverapi.DriverEncryptionConfig{}	×
295	keys, tags = c.getKeys(subsysIPSec)	×
296	drvEnc.Keys = keys	×
297	drvEnc.Tags = tags	×
298		×
299	c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {	×
300	err := driver.DiscoverNew(discoverapi.EncryptionKeysConfig, drvEnc)	×
301	if err != nil {	×
302	logrus.Warnf("Failed to set datapath keys in driver %s: %v", name, err)	×
303	}	×
304	return false	×
305	})
306
307	c.WalkNetworks(joinCluster)	×
308		×
309	return nil	×
310	}
311
312	func (c *controller) agentJoin(remote string) error {	×
313	agent := c.getAgent()	×
314	if agent == nil {	×
315	return nil	×
316	}	×
317	return agent.networkDB.Join([]string{remote})	×
318	}
319
320	func (c *controller) agentDriverNotify(d driverapi.Driver) {	76✔
321	agent := c.getAgent()	76✔
322	if agent == nil {	152✔
323	return	76✔
324	}	76✔
325
326	d.DiscoverNew(discoverapi.NodeDiscovery, discoverapi.NodeDiscoveryData{	×
327	Address: agent.advertiseAddr,	×
328	BindAddress: agent.bindAddr,	×
329	Self: true,	×
330	})	×
331		×
332	drvEnc := discoverapi.DriverEncryptionConfig{}	×
333	keys, tags := c.getKeys(subsysIPSec)	×
334	drvEnc.Keys = keys	×
335	drvEnc.Tags = tags	×
336		×
337	c.drvRegistry.WalkDrivers(func(name string, driver driverapi.Driver, capability driverapi.Capability) bool {	×
338	err := driver.DiscoverNew(discoverapi.EncryptionKeysConfig, drvEnc)	×
339	if err != nil {	×
340	logrus.Warnf("Failed to set datapath keys in driver %s: %v", name, err)	×
341	}	×
342	return false	×
343	})
344
345	}
346
347	func (c *controller) agentClose() {	×
348	// Acquire current agent instance and reset its pointer	×
349	// then run closing functions	×
350	c.Lock()	×
351	agent := c.agent	×
352	c.agent = nil	×
353	c.Unlock()	×
354		×
355	if agent == nil {	×
356	return	×
357	}	×
358
359	for _, cancelFuncs := range agent.driverCancelFuncs {	×
360	for _, cancel := range cancelFuncs {	×
361	cancel()	×
362	}	×
363	}
364
365	agent.epTblCancel()	×
366		×
367	agent.networkDB.Close()	×
368	}
369
370	func (n *network) isClusterEligible() bool {	454✔
371	if n.driverScope() != datastore.GlobalScope {	908✔
372	return false	454✔
373	}	454✔
374	return n.getController().getAgent() != nil	×
375	}
376
377	func (n *network) joinCluster() error {	42✔
378	if !n.isClusterEligible() {	84✔
379	return nil	42✔
380	}	42✔
381
382	agent := n.getController().getAgent()	×
383	if agent == nil {	×
384	return nil	×
385	}	×
386
387	return agent.networkDB.JoinNetwork(n.ID())	×
388	}
389
390	func (n *network) leaveCluster() error {	32✔
391	if !n.isClusterEligible() {	64✔
392	return nil	32✔
393	}	32✔
394
395	agent := n.getController().getAgent()	×
396	if agent == nil {	×
397	return nil	×
398	}	×
399
400	return agent.networkDB.LeaveNetwork(n.ID())	×
401	}
402
403	func (ep *endpoint) addDriverInfoToCluster() error {	102✔
404	n := ep.getNetwork()	102✔
405	if !n.isClusterEligible() {	204✔
406	return nil	102✔
407	}	102✔
408	if ep.joinInfo == nil {	×
409	return nil	×
410	}	×
411
412	agent := n.getController().getAgent()	×
413	if agent == nil {	×
414	return nil	×
415	}	×
416
417	for _, te := range ep.joinInfo.driverTableEntries {	×
418	if err := agent.networkDB.CreateEntry(te.tableName, n.ID(), te.key, te.value); err != nil {	×
419	return err	×
420	}	×
421	}
422	return nil	×
423	}
424
425	func (ep *endpoint) deleteDriverInfoFromCluster() error {	102✔
426	n := ep.getNetwork()	102✔
427	if !n.isClusterEligible() {	204✔
428	return nil	102✔
429	}	102✔
430	if ep.joinInfo == nil {	×
431	return nil	×
432	}	×
433
434	agent := n.getController().getAgent()	×
435	if agent == nil {	×
436	return nil	×
437	}	×
438
439	for _, te := range ep.joinInfo.driverTableEntries {	×
440	if err := agent.networkDB.DeleteEntry(te.tableName, n.ID(), te.key); err != nil {	×
441	return err	×
442	}	×
443	}
444	return nil	×
445	}
446
447	func (ep *endpoint) addServiceInfoToCluster() error {	×
448	n := ep.getNetwork()	×
449	if !n.isClusterEligible() {	×
450	return nil	×
451	}	×
452
453	c := n.getController()	×
454	agent := c.getAgent()	×
455	if !ep.isAnonymous() && ep.Iface().Address() != nil {	×
456	var ingressPorts []*PortConfig	×
457	if ep.svcID != "" {	×
458	// Gossip ingress ports only in ingress network.	×
459	if n.ingress {	×
460	ingressPorts = ep.ingressPorts	×
461	}	×
462
463	if err := c.addServiceBinding(ep.svcName, ep.svcID, n.ID(), ep.ID(), ep.virtualIP, ingressPorts, ep.svcAliases, ep.Iface().Address().IP); err != nil {	×
464	return err	×
465	}	×
466	}
467
468	buf, err := proto.Marshal(&EndpointRecord{	×
469	Name: ep.Name(),	×
470	ServiceName: ep.svcName,	×
471	ServiceID: ep.svcID,	×
472	VirtualIP: ep.virtualIP.String(),	×
473	IngressPorts: ingressPorts,	×
474	Aliases: ep.svcAliases,	×
475	TaskAliases: ep.myAliases,	×
476	EndpointIP: ep.Iface().Address().IP.String(),	×
477	})	×
478		×
479	if err != nil {	×
480	return err	×
481	}	×
482
483	if agent != nil {	×
484	if err := agent.networkDB.CreateEntry("endpoint_table", n.ID(), ep.ID(), buf); err != nil {	×
485	return err	×
486	}	×
487	}
488	}
489
490	return nil	×
491	}
492
493	func (ep *endpoint) deleteServiceInfoFromCluster() error {	102✔
494	n := ep.getNetwork()	102✔
495	if !n.isClusterEligible() {	204✔
496	return nil	102✔
497	}	102✔
498
499	c := n.getController()	×
500	agent := c.getAgent()	×
501		×
502	if !ep.isAnonymous() {	×
503	if ep.svcID != "" && ep.Iface().Address() != nil {	×
504	var ingressPorts []*PortConfig	×
505	if n.ingress {	×
506	ingressPorts = ep.ingressPorts	×
507	}	×
508
509	if err := c.rmServiceBinding(ep.svcName, ep.svcID, n.ID(), ep.ID(), ep.virtualIP, ingressPorts, ep.svcAliases, ep.Iface().Address().IP); err != nil {	×
510	return err	×
511	}	×
512	}
513	if agent != nil {	×
514	if err := agent.networkDB.DeleteEntry("endpoint_table", n.ID(), ep.ID()); err != nil {	×
515	return err	×
516	}	×
517	}
518	}
519	return nil	×
520	}
521
522	func (n *network) addDriverWatches() {	42✔
523	if !n.isClusterEligible() {	84✔
524	return	42✔
525	}	42✔
526
527	c := n.getController()	×
528	agent := c.getAgent()	×
529	if agent == nil {	×
530	return	×
531	}	×
532	for _, tableName := range n.driverTables {	×
533	ch, cancel := agent.networkDB.Watch(tableName, n.ID(), "")	×
534	agent.driverCancelFuncs[n.ID()] = append(agent.driverCancelFuncs[n.ID()], cancel)	×
535		×
536	go c.handleTableEvents(ch, n.handleDriverTableEvent)	×
537	d, err := n.driver(false)	×
538	if err != nil {	×
539	logrus.Errorf("Could not resolve driver %s while walking driver tabl: %v", n.networkType, err)	×
540	return	×
541	}	×
542
543	agent.networkDB.WalkTable(tableName, func(nid, key string, value []byte) bool {	×
544	if nid == n.ID() {	×
545	d.EventNotify(driverapi.Create, nid, tableName, key, value)	×
546	}	×
547
548	return false	×
549	})
550	}
551	}
552
553	func (n *network) cancelDriverWatches() {	32✔
554	if !n.isClusterEligible() {	64✔
555	return	32✔
556	}	32✔
557
558	c := n.getController()	×
559	c.Lock()	×
560	cancelFuncs := c.agent.driverCancelFuncs[n.ID()]	×
561	delete(c.agent.driverCancelFuncs, n.ID())	×
562	c.Unlock()	×
563		×
564	for _, cancel := range cancelFuncs {	×
565	cancel()	×
566	}	×
567	}
568
569	func (c *controller) handleTableEvents(ch chan events.Event, fn func(events.Event)) {	×
570	for {	×
571	select {	×
572	case ev, ok := <-ch:	×
573	if !ok {	×
574	return	×
575	}	×
576
577	fn(ev)	×
578	}
579	}
580	}
581
582	func (n *network) handleDriverTableEvent(ev events.Event) {	×
583	d, err := n.driver(false)	×
584	if err != nil {	×
585	logrus.Errorf("Could not resolve driver %s while handling driver table event: %v", n.networkType, err)	×
586	return	×
587	}	×
588
589	var (	×
590	etype driverapi.EventType	×
591	tname string	×
592	key string	×
593	value []byte	×
594	)	×
595		×
596	switch event := ev.(type) {	×
597	case networkdb.CreateEvent:	×
598	tname = event.Table	×
599	key = event.Key	×
600	value = event.Value	×
601	etype = driverapi.Create	×
602	case networkdb.DeleteEvent:	×
603	tname = event.Table	×
604	key = event.Key	×
605	value = event.Value	×
606	etype = driverapi.Delete	×
607	case networkdb.UpdateEvent:	×
608	tname = event.Table	×
609	key = event.Key	×
610	value = event.Value	×
611	etype = driverapi.Delete	×
612	}
613
614	d.EventNotify(etype, n.ID(), tname, key, value)	×
615	}
616
617	func (c *controller) handleEpTableEvent(ev events.Event) {	×
618	var (	×
619	nid string	×
620	eid string	×
621	value []byte	×
622	isAdd bool	×
623	epRec EndpointRecord	×
624	)	×
625		×
626	switch event := ev.(type) {	×
627	case networkdb.CreateEvent:	×
628	nid = event.NetworkID	×
629	eid = event.Key	×
630	value = event.Value	×
631	isAdd = true	×
632	case networkdb.DeleteEvent:	×
633	nid = event.NetworkID	×
634	eid = event.Key	×
635	value = event.Value	×
636	case networkdb.UpdateEvent:	×
637	logrus.Errorf("Unexpected update service table event = %#v", event)	×
638	}
639
640	nw, err := c.NetworkByID(nid)	×
641	if err != nil {	×
642	logrus.Errorf("Could not find network %s while handling service table event: %v", nid, err)	×
643	return	×
644	}	×
645	n := nw.(*network)	×
646		×
647	err = proto.Unmarshal(value, &epRec)	×
648	if err != nil {	×
649	logrus.Errorf("Failed to unmarshal service table value: %v", err)	×
650	return	×
651	}	×
652
653	name := epRec.Name	×
654	svcName := epRec.ServiceName	×
655	svcID := epRec.ServiceID	×
656	vip := net.ParseIP(epRec.VirtualIP)	×
657	ip := net.ParseIP(epRec.EndpointIP)	×
658	ingressPorts := epRec.IngressPorts	×
659	aliases := epRec.Aliases	×
660	taskaliases := epRec.TaskAliases	×
661		×
662	if name == "" \|\| ip == nil {	×
663	logrus.Errorf("Invalid endpoint name/ip received while handling service table event %s", value)	×
664	return	×
665	}	×
666
667	if isAdd {	×
668	if svcID != "" {	×
669	if err := c.addServiceBinding(svcName, svcID, nid, eid, vip, ingressPorts, aliases, ip); err != nil {	×
670	logrus.Errorf("Failed adding service binding for value %s: %v", value, err)	×
671	return	×
672	}	×
673	}
674
675	n.addSvcRecords(name, ip, nil, true)	×
676	for _, alias := range taskaliases {	×
677	n.addSvcRecords(alias, ip, nil, true)	×
678	}	×
679	} else {	×
680	if svcID != "" {	×
681	if err := c.rmServiceBinding(svcName, svcID, nid, eid, vip, ingressPorts, aliases, ip); err != nil {	×
682	logrus.Errorf("Failed adding service binding for value %s: %v", value, err)	×
683	return	×
684	}	×
685	}
686
687	n.deleteSvcRecords(name, ip, nil, true)	×
688	for _, alias := range taskaliases {	×
689	n.deleteSvcRecords(alias, ip, nil, true)	×
690	}	×
691	}
692	}

docker / libnetwork / #3248

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous