Merge pull request #1674 from sanimej/inspect

swarm mode network inspect should provide cluser-wide task details

libnetwork/agent.go

@@ -44,6 +44,8 @@ type agent struct {
 	sync.Mutex
 }
 
+const libnetworkEPTable = "endpoint_table"
+
 func getBindAddr(ifaceName string) (string, error) {
 	iface, err := net.InterfaceByName(ifaceName)
 	if err != nil {
@@ -285,7 +287,7 @@ func (c *controller) agentInit(listenAddr, bindAddrOrInterface, advertiseAddr st
 		return err
 	}
 
-	ch, cancel := nDB.Watch("endpoint_table", "", "")
+	ch, cancel := nDB.Watch(libnetworkEPTable, "", "")
 	nodeCh, cancel := nDB.Watch(networkdb.NodeTable, "", "")
 
 	c.Lock()
@@ -385,6 +387,111 @@ func (c *controller) agentClose() {
 	agent.networkDB.Close()
 }
 
+// Task has the backend container details
+type Task struct {
+	Name       string
+	EndpointID string
+	EndpointIP string
+	Info       map[string]string
+}
+
+// ServiceInfo has service specific details along with the list of backend tasks
+type ServiceInfo struct {
+	VIP          string
+	LocalLBIndex int
+	Tasks        []Task
+	Ports        []string
+}
+
+type epRecord struct {
+	ep      EndpointRecord
+	info    map[string]string
+	lbIndex int
+}
+
+func (n *network) Services() map[string]ServiceInfo {
+	eps := make(map[string]epRecord)
+
+	if !n.isClusterEligible() {
+		return nil
+	}
+	agent := n.getController().getAgent()
+	if agent == nil {
+		return nil
+	}
+
+	// Walk through libnetworkEPTable and fetch the driver agnostic endpoint info
+	entries := agent.networkDB.GetTableByNetwork(libnetworkEPTable, n.id)
+	for eid, value := range entries {
+		var epRec EndpointRecord
+		nid := n.ID()
+		if err := proto.Unmarshal(value.([]byte), &epRec); err != nil {
+			logrus.Errorf("Unmarshal of libnetworkEPTable failed for endpoint %s in network %s, %v", eid, nid, err)
+			continue
+		}
+		i := n.getController().getLBIndex(epRec.ServiceID, nid, epRec.IngressPorts)
+		eps[eid] = epRecord{
+			ep:      epRec,
+			lbIndex: i,
+		}
+	}
+
+	// Walk through the driver's tables, have the driver decode the entries
+	// and return the tuple {ep ID, value}. value is a string that coveys
+	// relevant info about the endpoint.
+	d, err := n.driver(true)
+	if err != nil {
+		logrus.Errorf("Could not resolve driver for network %s/%s while fetching services: %v", n.networkType, n.ID(), err)
+		return nil
+	}
+	for _, table := range n.driverTables {
+		if table.objType != driverapi.EndpointObject {
+			continue
+		}
+		entries := agent.networkDB.GetTableByNetwork(table.name, n.id)
+		for key, value := range entries {
+			epID, info := d.DecodeTableEntry(table.name, key, value.([]byte))
+			if ep, ok := eps[epID]; !ok {
+				logrus.Errorf("Inconsistent driver and libnetwork state for endpoint %s", epID)
+			} else {
+				ep.info = info
+				eps[epID] = ep
+			}
+		}
+	}
+
+	// group the endpoints into a map keyed by the service name
+	sinfo := make(map[string]ServiceInfo)
+	for ep, epr := range eps {
+		var (
+			s  ServiceInfo
+			ok bool
+		)
+		if s, ok = sinfo[epr.ep.ServiceName]; !ok {
+			s = ServiceInfo{
+				VIP:          epr.ep.VirtualIP,
+				LocalLBIndex: epr.lbIndex,
+			}
+		}
+		ports := []string{}
+		if s.Ports == nil {
+			for _, port := range epr.ep.IngressPorts {
+				p := fmt.Sprintf("Target: %d, Publish: %d", port.TargetPort, port.PublishedPort)
+				ports = append(ports, p)
+			}
+			s.Ports = ports
+		}
+		s.Tasks = append(s.Tasks, Task{
+			Name:       epr.ep.Name,
+			EndpointID: ep,
+			EndpointIP: epr.ep.EndpointIP,
+			Info:       epr.info,
+		})
+		sinfo[epr.ep.ServiceName] = s
+	}
+	return sinfo
+}
+
 func (n *network) isClusterEligible() bool {
 	if n.driverScope() != datastore.GlobalScope {
 		return false
@@ -508,7 +615,7 @@ func (ep *endpoint) addServiceInfoToCluster() error {
 	}
 
 	if agent != nil {
-		if err := agent.networkDB.CreateEntry("endpoint_table", n.ID(), ep.ID(), buf); err != nil {
+		if err := agent.networkDB.CreateEntry(libnetworkEPTable, n.ID(), ep.ID(), buf); err != nil {
 			return err
 		}
 	}
@@ -541,7 +648,7 @@ func (ep *endpoint) deleteServiceInfoFromCluster() error {
 	}
 
 	if agent != nil {
-		if err := agent.networkDB.DeleteEntry("endpoint_table", n.ID(), ep.ID()); err != nil {
+		if err := agent.networkDB.DeleteEntry(libnetworkEPTable, n.ID(), ep.ID()); err != nil {
 			return err
 		}
 	}
@@ -559,8 +666,8 @@ func (n *network) addDriverWatches() {
 	if agent == nil {
 		return
 	}
-	for _, tableName := range n.driverTables {
-		ch, cancel := agent.networkDB.Watch(tableName, n.ID(), "")
+	for _, table := range n.driverTables {
+		ch, cancel := agent.networkDB.Watch(table.name, n.ID(), "")
 		agent.Lock()
 		agent.driverCancelFuncs[n.ID()] = append(agent.driverCancelFuncs[n.ID()], cancel)
 		agent.Unlock()
@@ -571,9 +678,9 @@ func (n *network) addDriverWatches() {
 			return
 		}
 
-		agent.networkDB.WalkTable(tableName, func(nid, key string, value []byte) bool {
+		agent.networkDB.WalkTable(table.name, func(nid, key string, value []byte) bool {
 			if nid == n.ID() {
-				d.EventNotify(driverapi.Create, nid, tableName, key, value)
+				d.EventNotify(driverapi.Create, nid, table.name, key, value)
 			}
 
 			return false

libnetwork/driverapi/driverapi.go

@@ -72,6 +72,16 @@ type Driver interface {
 	// only invoked for the global scope driver.
 	EventNotify(event EventType, nid string, tableName string, key string, value []byte)
 
+	// DecodeTableEntry passes the driver a key, value pair from table it registered
+	// with libnetwork. Driver should return {object ID, map[string]string} tuple.
+	// If DecodeTableEntry is called for a table associated with NetworkObject or
+	// EndpointObject the return object ID should be the network id or endppoint id
+	// associated with that entry. map should have information about the object that
+	// can be presented to the user.
+	// For exampe: overlay driver returns the VTEP IP of the host that has the endpoint
+	// which is shown in 'network inspect --verbose'
+	DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string)
+
 	// Type returns the type of this driver, the network type this driver manages
 	Type() string
 
@@ -84,7 +94,7 @@ type Driver interface {
 type NetworkInfo interface {
 	// TableEventRegister registers driver interest in a given
 	// table name.
-	TableEventRegister(tableName string) error
+	TableEventRegister(tableName string, objType ObjectType) error
 }
 
 // InterfaceInfo provides a go interface for drivers to retrive
@@ -175,3 +185,28 @@ const (
 	// Delete event is generated when a table entry is deleted.
 	Delete
 )
+
+// ObjectType represents the type of object driver wants to store in libnetwork's networkDB
+type ObjectType int
+
+const (
+	// EndpointObject should be set for libnetwork endpoint object related data
+	EndpointObject ObjectType = 1 + iota
+	// NetworkObject should be set for libnetwork network object related data
+	NetworkObject
+	// OpaqueObject is for driver specific data with no corresponding libnetwork object
+	OpaqueObject
+)
+
+// IsValidType validates the passed in type against the valid object types
+func IsValidType(objType ObjectType) bool {
+	switch objType {
+	case EndpointObject:
+		fallthrough
+	case NetworkObject:
+		fallthrough
+	case OpaqueObject:
+		return true
+	}
+	return false
+}

libnetwork/drivers/bridge/bridge.go

@@ -575,6 +575,10 @@ func (d *driver) NetworkFree(id string) error {
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 // Create a new network using bridge plugin
 func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	if len(ipV4Data) == 0 || ipV4Data[0].Pool.String() == "0.0.0.0/0" {

libnetwork/drivers/host/host.go

@@ -35,6 +35,10 @@ func (d *driver) NetworkFree(id string) error {
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	d.Lock()
 	defer d.Unlock()

libnetwork/drivers/ipvlan/ipvlan.go

@@ -108,3 +108,7 @@ func (d *driver) DiscoverDelete(dType discoverapi.DiscoveryType, data interface{
 
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
+
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}

libnetwork/drivers/macvlan/macvlan.go

@@ -110,3 +110,7 @@ func (d *driver) DiscoverDelete(dType discoverapi.DiscoveryType, data interface{
 
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
+
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}

libnetwork/drivers/null/null.go

@@ -35,6 +35,10 @@ func (d *driver) NetworkFree(id string) error {
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	d.Lock()
 	defer d.Unlock()

libnetwork/drivers/overlay/joinleave.go

@@ -145,6 +145,23 @@ func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo,
 	return nil
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	if tablename != ovPeerTable {
+		logrus.Errorf("DecodeTableEntry: unexpected table name %s", tablename)
+		return "", nil
+	}
+
+	var peer PeerRecord
+	if err := proto.Unmarshal(value, &peer); err != nil {
+		logrus.Errorf("DecodeTableEntry: failed to unmarshal peer record for key %s: %v", key, err)
+		return "", nil
+	}
+
+	return key, map[string]string{
+		"Host IP": peer.TunnelEndpointIP,
+	}
+}
+
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 	if tableName != ovPeerTable {
 		logrus.Errorf("Unexpected table notification for table %s received", tableName)

libnetwork/drivers/overlay/ov_network.go

@@ -159,7 +159,7 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d
 	}
 
 	if nInfo != nil {
-		if err := nInfo.TableEventRegister(ovPeerTable); err != nil {
+		if err := nInfo.TableEventRegister(ovPeerTable, driverapi.EndpointObject); err != nil {
 			return err
 		}
 	}

libnetwork/drivers/overlay/ovmanager/ovmanager.go

@@ -199,6 +199,10 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) DeleteNetwork(nid string) error {
 	return types.NotImplementedErrorf("not implemented")
 }

libnetwork/drivers/remote/driver.go

@@ -116,6 +116,10 @@ func (d *driver) NetworkFree(id string) error {
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) CreateNetwork(id string, options map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	create := &api.CreateNetworkRequest{
 		NetworkID: id,

libnetwork/drivers/solaris/bridge/bridge.go

@@ -175,6 +175,10 @@ func (d *driver) NetworkFree(id string) error {
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	if len(ipV4Data) == 0 || ipV4Data[0].Pool.String() == "0.0.0.0/0" {
 		return types.BadRequestErrorf("ipv4 pool is empty")

libnetwork/drivers/solaris/overlay/joinleave.go

@@ -149,6 +149,10 @@ func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key stri
 	d.peerAdd(nid, eid, addr.IP, addr.Mask, mac, vtep, true)
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 // Leave method is invoked when a Sandbox detaches from an endpoint.
 func (d *driver) Leave(nid, eid string) error {
 	if err := validateID(nid, eid); err != nil {

libnetwork/drivers/solaris/overlay/ov_network.go

@@ -153,7 +153,7 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d
 	}
 
 	if nInfo != nil {
-		if err := nInfo.TableEventRegister(ovPeerTable); err != nil {
+		if err := nInfo.TableEventRegister(ovPeerTable, driverapi.EndpointObject); err != nil {
 			return err
 		}
 	}

libnetwork/drivers/solaris/overlay/ovmanager/ovmanager.go

@@ -199,6 +199,10 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func (d *driver) DeleteNetwork(nid string) error {
 	return types.NotImplementedErrorf("not implemented")
 }

libnetwork/drivers/windows/overlay/joinleave_windows.go

@@ -93,6 +93,10 @@ func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key stri
 	d.peerAdd(nid, eid, addr.IP, addr.Mask, mac, vtep, true)
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 // Leave method is invoked when a Sandbox detaches from an endpoint.
 func (d *driver) Leave(nid, eid string) error {
 	if err := validateID(nid, eid); err != nil {

libnetwork/drivers/windows/overlay/ov_network_windows.go

@@ -169,7 +169,7 @@ func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo d
 	n.interfaceName = interfaceName
 
 	if nInfo != nil {
-		if err := nInfo.TableEventRegister(ovPeerTable); err != nil {
+		if err := nInfo.TableEventRegister(ovPeerTable, driverapi.EndpointObject); err != nil {
 			return err
 		}
 	}

libnetwork/drivers/windows/windows.go

@@ -183,6 +183,10 @@ func (c *networkConfiguration) processIPAM(id string, ipamV4Data, ipamV6Data []d
 func (d *driver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (d *driver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 // Create a new network
 func (d *driver) CreateNetwork(id string, option map[string]interface{}, nInfo driverapi.NetworkInfo, ipV4Data, ipV6Data []driverapi.IPAMData) error {
 	if _, err := d.getNetwork(id); err == nil {

libnetwork/drvregistry/drvregistry_test.go

@@ -91,6 +91,10 @@ func (m *mockDriver) NetworkFree(id string) error {
 func (m *mockDriver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
 
+func (m *mockDriver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}
+
 func getNew(t *testing.T) *DrvRegistry {
 	reg, err := New(nil, nil, nil, nil, nil)
 	if err != nil {

libnetwork/libnetwork_internal_test.go

@@ -564,3 +564,7 @@ func (b *badDriver) NetworkFree(id string) error {
 
 func (b *badDriver) EventNotify(etype driverapi.EventType, nid, tableName, key string, value []byte) {
 }
+
+func (b *badDriver) DecodeTableEntry(tablename string, key string, value []byte) (string, map[string]string) {
+	return "", nil
+}

libnetwork/network.go

@@ -74,6 +74,9 @@ type NetworkInfo interface {
 	// gossip cluster. For non-dynamic overlay networks and bridge networks it returns an
 	// empty slice
 	Peers() []networkdb.PeerInfo
+	//Services returns a map of services keyed by the service name with the details
+	//of all the tasks that belong to the service. Applicable only in swarm mode.
+	Services() map[string]ServiceInfo
 }
 
 // EndpointWalker is a client provided function which will be used to walk the Endpoints.
@@ -108,6 +111,11 @@ type servicePorts struct {
 	target   []serviceTarget
 }
 
+type networkDBTable struct {
+	name    string
+	objType driverapi.ObjectType
+}
+
 // IpamConf contains all the ipam related configurations for a network
 type IpamConf struct {
 	// The master address pool for containers and network interfaces
@@ -208,7 +216,7 @@ type network struct {
 	attachable   bool
 	inDelete     bool
 	ingress      bool
-	driverTables []string
+	driverTables []networkDBTable
 	dynamic      bool
 	sync.Mutex
 }
@@ -1607,11 +1615,18 @@ func (n *network) Labels() map[string]string {
 	return lbls
 }
 
-func (n *network) TableEventRegister(tableName string) error {
+func (n *network) TableEventRegister(tableName string, objType driverapi.ObjectType) error {
+	if !driverapi.IsValidType(objType) {
+		return fmt.Errorf("invalid object type %v in registering table, %s", objType, tableName)
+	}
+
+	t := networkDBTable{
+		name:    tableName,
+		objType: objType,
+	}
 	n.Lock()
 	defer n.Unlock()
-
-	n.driverTables = append(n.driverTables, tableName)
+	n.driverTables = append(n.driverTables, t)
 	return nil
 }
 

libnetwork/networkdb/networkdb.go

@@ -307,6 +307,22 @@ func (nDB *NetworkDB) UpdateEntry(tname, nid, key string, value []byte) error {
 	return nil
 }
 
+// GetTableByNetwork walks the networkdb by the give table and network id and
+// returns a map of keys and values
+func (nDB *NetworkDB) GetTableByNetwork(tname, nid string) map[string]interface{} {
+	entries := make(map[string]interface{})
+	nDB.indexes[byTable].WalkPrefix(fmt.Sprintf("/%s/%s", tname, nid), func(k string, v interface{}) bool {
+		entry := v.(*entry)
+		if entry.deleting {
+			return false
+		}
+		key := k[strings.LastIndex(k, "/")+1:]
+		entries[key] = entry.value
+		return false
+	})
+	return entries
+}
+
 // DeleteEntry deletes a table entry in NetworkDB for given (network,
 // table, key) tuple and if the NetworkDB is part of the cluster
 // propagates this event to the cluster.

libnetwork/service_common.go

@@ -18,6 +18,26 @@ func newService(name string, id string, ingressPorts []*PortConfig, aliases []st
 	}
 }
 
+func (c *controller) getLBIndex(sid, nid string, ingressPorts []*PortConfig) int {
+	skey := serviceKey{
+		id:    sid,
+		ports: portConfigs(ingressPorts).String(),
+	}
+	c.Lock()
+	s, ok := c.serviceBindings[skey]
+	c.Unlock()
+
+	if !ok {
+		return 0
+	}
+
+	s.Lock()
+	lb := s.loadBalancers[nid]
+	s.Unlock()
+
+	return int(lb.fwMark)
+}
+
 func (c *controller) cleanupServiceBindings(cleanupNID string) {
 	var cleanupFuncs []func()