Merge issues/1505-fast-datapath-weave-status' (PR #1525)

Fast datapath weave status

prog/weaver/http.go

@@ -161,7 +161,12 @@ func HandleHTTP(muxRouter *mux.Router, version string, router *weave.Router, all
 	}
 	muxRouter.Methods("GET").Path("/report").Headers("Accept", "application/json").HandlerFunc(
 		func(w http.ResponseWriter, r *http.Request) {
-			json, _ := json.MarshalIndent(status(), "", "    ")
+			json, err := json.MarshalIndent(status(), "", "    ")
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				Log.Error("Error during report marshalling: ", err)
+				return
+			}
 			w.Header().Set("Content-Type", "application/json")
 			w.Write(json)
 		})

router/fastdp.go

@@ -2,6 +2,7 @@ package router
 
 import (
 	"encoding/binary"
+	"encoding/json"
 	"fmt"
 	"net"
 	"sync"
@@ -321,6 +322,77 @@ func (fastDatapathOverlay) AddFeaturesTo(features map[string]string) {
 	// OverlaySwitch.
 }
 
+type FlowStatus odp.FlowInfo
+
+func (flowStatus *FlowStatus) MarshalJSON() ([]byte, error) {
+	type jsonFlowStatus struct {
+		FlowKeys []string
+		Actions  []string
+		Packets  uint64
+		Bytes    uint64
+		Used     uint64
+	}
+
+	flowKeys := make([]string, 0, len(flowStatus.FlowKeys))
+	for _, flowKey := range flowStatus.FlowKeys {
+		if !flowKey.Ignored() {
+			flowKeys = append(flowKeys, fmt.Sprint(flowKey))
+		}
+	}
+
+	actions := make([]string, 0, len(flowStatus.Actions))
+	for _, action := range flowStatus.Actions {
+		actions = append(actions, fmt.Sprint(action))
+	}
+
+	return json.Marshal(&jsonFlowStatus{flowKeys, actions,
+		flowStatus.Packets, flowStatus.Bytes, flowStatus.Used})
+}
+
+type VportStatus odp.Vport
+
+func (vport *VportStatus) MarshalJSON() ([]byte, error) {
+	type jsonVportStatus struct {
+		ID       odp.VportID
+		Name     string
+		TypeName string
+	}
+
+	return json.Marshal(&jsonVportStatus{vport.ID, vport.Spec.Name(),
+		vport.Spec.TypeName()})
+}
+
+func (fastdp fastDatapathOverlay) Diagnostics() interface{} {
+	lock := fastdp.startLock()
+	defer lock.unlock()
+
+	vports, err := fastdp.dp.EnumerateVports()
+	if err != nil {
+		log.Warn(err)
+	}
+	vportStatuses := make([]VportStatus, 0, len(vports))
+	for _, vport := range vports {
+		vportStatuses = append(vportStatuses, VportStatus(vport))
+	}
+
+	flows, err := fastdp.dp.EnumerateFlows()
+	if err != nil {
+		log.Warn(err)
+	}
+	flowStatuses := make([]FlowStatus, 0, len(flows))
+	for _, flow := range flows {
+		flowStatuses = append(flowStatuses, FlowStatus(flow))
+	}
+
+	return struct {
+		Vports []VportStatus
+		Flows  []FlowStatus
+	}{
+		vportStatuses,
+		flowStatuses,
+	}
+}
+
 func (fastdp fastDatapathOverlay) StartConsumingPackets(localPeer *Peer,
 	peers *Peers, consumer OverlayConsumer) error {
 	fastdp.lock.Lock()
@@ -662,6 +734,10 @@ func (fwd *fastDatapathForwarder) ControlMessage(tag byte, msg []byte) {
 	}
 }
 
+func (fwd *fastDatapathForwarder) DisplayName() string {
+	return "fastdp"
+}
+
 func (fwd *fastDatapathForwarder) handleHeartbeatAck() {
 	log.Debug(fwd.logPrefix(), "handleHeartbeatAck")
 

router/overlay.go

@@ -21,6 +21,9 @@ type Overlay interface {
 
 	// Enhance a features map with overlay-related features
 	AddFeaturesTo(map[string]string)
+
+	// Obtain diagnostic information specific to the overlay
+	Diagnostics() interface{}
 }
 
 type ForwarderParams struct {
@@ -88,6 +91,9 @@ type OverlayForwarder interface {
 	// compatibility, and should always be
 	// ProtocolOverlayControlMessage for non-sleeve overlays.
 	ControlMessage(tag byte, msg []byte)
+
+	// User facing overlay name
+	DisplayName() string
 }
 
 type NullOverlay struct{}
@@ -129,3 +135,11 @@ func (NullOverlay) Stop() {
 
 func (NullOverlay) ControlMessage(byte, []byte) {
 }
+
+func (NullOverlay) DisplayName() string {
+	return "null"
+}
+
+func (NullOverlay) Diagnostics() interface{} {
+	return nil
+}

router/overlay_switch.go

@@ -35,6 +35,14 @@ func (osw *OverlaySwitch) AddFeaturesTo(features map[string]string) {
 	features["Overlays"] = strings.Join(osw.overlayNames, " ")
 }
 
+func (osw *OverlaySwitch) Diagnostics() interface{} {
+	diagnostics := make(map[string]interface{})
+	for name, overlay := range osw.overlays {
+		diagnostics[name] = overlay.Diagnostics()
+	}
+	return diagnostics
+}
+
 func (osw *OverlaySwitch) InvalidateRoutes() {
 	for _, overlay := range osw.overlays {
 		overlay.InvalidateRoutes()
@@ -393,3 +401,15 @@ func (fwd *overlaySwitchForwarder) ControlMessage(tag byte, msg []byte) {
 		subFwd.ControlMessage(msg[1], msg[2:])
 	}
 }
+
+func (fwd *overlaySwitchForwarder) DisplayName() string {
+	fwd.lock.Lock()
+	best := fwd.best
+	fwd.lock.Unlock()
+
+	if best == nil {
+		return "none"
+	}
+
+	return best.DisplayName()
+}

router/sleeve.go

@@ -133,6 +133,10 @@ func (*SleeveOverlay) AddFeaturesTo(map[string]string) {
 	// No features to be provided, to facilitate compatibility
 }
 
+func (*SleeveOverlay) Diagnostics() interface{} {
+	return nil
+}
+
 func (sleeve *SleeveOverlay) lookupForwarder(peer PeerName) *sleeveForwarder {
 	sleeve.lock.Lock()
 	defer sleeve.lock.Unlock()
@@ -593,6 +597,10 @@ func (fwd *sleeveForwarder) ControlMessage(tag byte, msg []byte) {
 	}
 }
 
+func (fwd *sleeveForwarder) DisplayName() string {
+	return "sleeve"
+}
+
 func (fwd *sleeveForwarder) Stop() {
 	fwd.sleeve.removeForwarder(fwd.remotePeer.Name, fwd)
 

router/status.go

@@ -1,6 +1,7 @@
 package router
 
 import (
+	"fmt"
 	"time"
 )
 
@@ -21,6 +22,7 @@ type Status struct {
 	BroadcastRoutes    []BroadcastRouteStatus
 	Connections        []LocalConnectionStatus
 	Targets            []string
+	OverlayDiagnostics interface{}
 }
 
 type MACStatus struct {
@@ -34,6 +36,7 @@ type PeerStatus struct {
 	Name        string
 	NickName    string
 	UID         PeerUID
+	ShortID     PeerShortID
 	Version     uint64
 	Connections []ConnectionStatus
 }
@@ -79,7 +82,8 @@ func NewStatus(router *Router) *Status {
 		NewUnicastRouteStatusSlice(router.Routes),
 		NewBroadcastRouteStatusSlice(router.Routes),
 		NewLocalConnectionStatusSlice(router.ConnectionMaker),
-		NewTargetSlice(router.ConnectionMaker)}
+		NewTargetSlice(router.ConnectionMaker),
+		router.Overlay.Diagnostics()}
 }
 
 func NewMACStatusSlice(cache *MacCache) []MACStatus {
@@ -119,6 +123,7 @@ func NewPeerStatusSlice(peers *Peers) []PeerStatus {
 			peer.Name.String(),
 			peer.NickName,
 			peer.UID,
+			peer.ShortID,
 			peer.Version,
 			connections})
 	})
@@ -170,7 +175,9 @@ func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus
 			if conn.Established() {
 				state = "established"
 			}
-			slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, conn.Remote().String()})
+			lc, _ := conn.(*LocalConnection)
+			info := fmt.Sprintf("%-6v %v", lc.forwarder.DisplayName(), conn.Remote())
+			slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, info})
 		}
 		for address, target := range cm.targets {
 			add := func(state, info string) {