Merge pull request #1539 from dvandra/OVS_SFlow_Metric

Adding OVS-SFlow Metric to Metadata

Author: safchain

Makefile

@@ -52,6 +52,7 @@ EASYJSON_FILES_TAG=\
 	flow/storage/elasticsearch/elasticsearch.go \
 	flow/storage/orientdb/orientdb.go \
 	graffiti/graph/elasticsearch.go \
+	sflow/sflowmetric.go \
 	topology/metrics.go \
 	topology/probes/netlink/route.go \
 	topology/probes/netlink/neighbor.go

analyzer/server.go

@@ -47,6 +47,7 @@ import (
 	"github.com/skydive-project/skydive/logging"
 	"github.com/skydive-project/skydive/packetinjector"
 	"github.com/skydive-project/skydive/probe"
+	"github.com/skydive-project/skydive/sflow"
 	"github.com/skydive-project/skydive/topology"
 	usertopology "github.com/skydive-project/skydive/topology/enhancers"
 	"github.com/skydive-project/skydive/topology/probes/netlink"
@@ -223,6 +224,7 @@ func NewServerFromConfig() (*Server, error) {
 	uiServer.AddGlobalVar("ui", config.Get("ui"))
 	uiServer.AddGlobalVar("flow-metric-keys", (&flow.FlowMetric{}).GetFieldKeys())
 	uiServer.AddGlobalVar("interface-metric-keys", (&topology.InterfaceMetric{}).GetFieldKeys())
+	uiServer.AddGlobalVar("sflow-metric-keys", (&sflow.SFMetric{}).GetFieldKeys())
 	uiServer.AddGlobalVar("probes", config.Get("analyzer.topology.probes"))
 
 	persistent, err := newGraphBackendFromConfig(etcdClient)
@@ -383,4 +385,5 @@ func init() {
 	graph.NodeMetadataDecoders["Neighbors"] = netlink.NeighborMetadataDecoder
 	graph.NodeMetadataDecoders["Metric"] = topology.InterfaceMetricMetadataDecoder
 	graph.NodeMetadataDecoders["LastUpdateMetric"] = topology.InterfaceMetricMetadataDecoder
+	graph.NodeMetadataDecoders["SFlow"] = sflow.SFMetadataDecoder
 }

api/client/gremlin.go

@@ -35,6 +35,7 @@ import (
 	"github.com/skydive-project/skydive/graffiti/graph"
 	"github.com/skydive-project/skydive/gremlin"
 	shttp "github.com/skydive-project/skydive/http"
+	"github.com/skydive-project/skydive/sflow"
 	"github.com/skydive-project/skydive/topology"
 	"github.com/skydive-project/skydive/topology/probes/socketinfo"
 )
@@ -178,6 +179,25 @@ func (g *GremlinQueryHelper) GetInterfaceMetrics(query interface{}) (map[string]
 	return result[0], nil
 }
 
+// GetSFlowMetrics from Gremlin query
+func (g *GremlinQueryHelper) GetSFlowMetrics(query interface{}) (map[string][]*sflow.SFMetric, error) {
+	data, err := g.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var result []map[string][]*sflow.SFMetric
+	if err := json.Unmarshal(data, &result); err != nil {
+		return nil, err
+	}
+
+	if len(result) == 0 {
+		return nil, nil
+	}
+
+	return result[0], nil
+}
+
 // GetFlowMetrics from Gremlin query
 func (g *GremlinQueryHelper) GetFlowMetrics(query interface{}) (map[string][]*flow.FlowMetric, error) {
 	data, err := g.Query(query)
@@ -227,6 +247,21 @@ func (g *GremlinQueryHelper) GetInterfaceMetric(query interface{}) (*topology.In
 	return &result, nil
 }
 
+// GetSFlowMetric from Gremlin query
+func (g *GremlinQueryHelper) GetSFlowMetric(query interface{}) (*sflow.SFMetric, error) {
+	data, err := g.Query(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var result sflow.SFMetric
+	if err := json.Unmarshal(data, &result); err != nil {
+		return nil, err
+	}
+
+	return &result, nil
+}
+
 // GetSockets from the Gremlin query
 func (g *GremlinQueryHelper) GetSockets(query interface{}) (sockets map[string][]*socketinfo.ConnectionInfo, err error) {
 	data, err := g.Query(query)

flow/probes/ovssflow.go

@@ -56,6 +56,7 @@ type OvsSFlowProbesHandler struct {
 	probes       map[string]OvsSFlowProbe
 	probesLock   common.RWMutex
 	Graph        *graph.Graph
+	Node         *graph.Node
 	fpta         *FlowProbeTableAllocator
 	ovsClient    *ovsdb.OvsClient
 	allocator    *sflow.AgentAllocator
@@ -178,7 +179,7 @@ func (o *OvsSFlowProbesHandler) UnregisterSFlowProbeFromBridge(bridgeUUID string
 }
 
 // RegisterProbeOnBridge registers a new probe on the OVS bridge
-func (o *OvsSFlowProbesHandler) RegisterProbeOnBridge(bridgeUUID string, tid string, capture *types.Capture) error {
+func (o *OvsSFlowProbesHandler) RegisterProbeOnBridge(bridgeUUID string, tid string, capture *types.Capture, n *graph.Node) error {
 	headerSize := flow.DefaultCaptureLength
 	if capture.HeaderSize != 0 {
 		headerSize = uint32(capture.HeaderSize)
@@ -192,7 +193,7 @@ func (o *OvsSFlowProbesHandler) RegisterProbeOnBridge(bridgeUUID string, tid str
 		Interface:  "lo",
 		HeaderSize: headerSize,
 		Sampling:   1,
-		Polling:    0,
+		Polling:    10,
 		flowTable:  ft,
 	}
 
@@ -202,7 +203,7 @@ func (o *OvsSFlowProbesHandler) RegisterProbeOnBridge(bridgeUUID string, tid str
 	}
 
 	addr := common.ServiceAddress{Addr: address, Port: 0}
-	agent, err := o.allocator.Alloc(bridgeUUID, probe.flowTable, capture.BPFFilter, headerSize, &addr)
+	agent, err := o.allocator.Alloc(bridgeUUID, probe.flowTable, capture.BPFFilter, headerSize, &addr, n, o.Graph)
 	if err != nil && err != sflow.ErrAgentAlreadyAllocated {
 		return err
 	}
@@ -231,7 +232,7 @@ func (o *OvsSFlowProbesHandler) registerProbe(n *graph.Node, capture *types.Capt
 
 	if isOvsBridge(n) {
 		if uuid, _ := n.GetFieldString("UUID"); uuid != "" {
-			if err := o.RegisterProbeOnBridge(uuid, tid, capture); err != nil {
+			if err := o.RegisterProbeOnBridge(uuid, tid, capture, n); err != nil {
 				return err
 			}
 			go e.OnStarted()

flow/probes/sflow.go

@@ -107,7 +107,7 @@ func (d *SFlowProbesHandler) registerProbe(n *graph.Node, capture *types.Capture
 	ft := d.fpta.Alloc(tid, opts)
 
 	addr := common.ServiceAddress{Addr: address, Port: capture.Port}
-	if _, err := d.allocator.Alloc(tid, ft, capture.BPFFilter, headerSize, &addr); err != nil {
+	if _, err := d.allocator.Alloc(tid, ft, capture.BPFFilter, headerSize, &addr, n, d.Graph); err != nil {
 		return err
 	}
 

gremlin/query.go

@@ -188,8 +188,8 @@ func (q QueryString) BothV() QueryString {
 }
 
 // Metrics append a Metrics() operation to query
-func (q QueryString) Metrics() QueryString {
-	return q.newQueryString("Metrics")
+func (q QueryString) Metrics(key ...interface{}) QueryString {
+	return q.newQueryString("Metrics", key...)
 }
 
 // Sum append a Sum() operation to query

gremlin/traversal/metrics.go

@@ -43,6 +43,7 @@ type MetricsTraversalExtension struct {
 // MetricsGremlinTraversalStep describes the Metrics gremlin traversal step
 type MetricsGremlinTraversalStep struct {
 	traversal.GremlinTraversalContext
+	key string
 }
 
 // NewMetricsTraversalExtension returns a new graph traversal extension
@@ -65,16 +66,38 @@ func (e *MetricsTraversalExtension) ScanIdent(s string) (traversal.Token, bool)
 func (e *MetricsTraversalExtension) ParseStep(t traversal.Token, p traversal.GremlinTraversalContext) (traversal.GremlinTraversalStep, error) {
 	switch t {
 	case e.MetricsToken:
-		return &MetricsGremlinTraversalStep{GremlinTraversalContext: p}, nil
+	default:
+		return nil, nil
 	}
-	return nil, nil
+	var key string
+	switch len(p.Params) {
+	case 0:
+		key = "LastUpdateMetric"
+	case 1:
+		k, ok := p.Params[0].(string)
+		if !ok {
+			return nil, errors.New("Metrics parameter have to be a string")
+		}
+		switch k {
+		case "LastUpdateMetric":
+			key = k
+		case "SFlow.LastUpdateMetric", "SFlow":
+			key = "SFlow.LastUpdateMetric"
+		default:
+			return nil, fmt.Errorf("Metric field unknown : %v", p.Params)
+		}
+	default:
+		return nil, fmt.Errorf("Metrics accepts one parameter : %v", p.Params)
+	}
+
+	return &MetricsGremlinTraversalStep{GremlinTraversalContext: p, key: key}, nil
 }
 
 // Exec executes the metrics step
 func (s *MetricsGremlinTraversalStep) Exec(last traversal.GraphTraversalStep) (traversal.GraphTraversalStep, error) {
 	switch tv := last.(type) {
 	case *traversal.GraphTraversalV:
-		return InterfaceMetrics(s.StepContext, tv), nil
+		return InterfaceMetrics(s.StepContext, tv, s.key), nil
 	case *FlowTraversalStep:
 		return tv.FlowMetrics(s.StepContext), nil
 	}

gremlin/traversal/topology.go

@@ -29,17 +29,19 @@ import (
 	"github.com/skydive-project/skydive/common"
 	"github.com/skydive-project/skydive/graffiti/graph"
 	"github.com/skydive-project/skydive/graffiti/graph/traversal"
-	"github.com/skydive-project/skydive/topology"
 	"github.com/skydive-project/skydive/topology/probes/socketinfo"
 )
 
 // InterfaceMetrics returns a Metrics step from interface metric metadata
-func InterfaceMetrics(ctx traversal.StepContext, tv *traversal.GraphTraversalV) *MetricsTraversalStep {
+func InterfaceMetrics(ctx traversal.StepContext, tv *traversal.GraphTraversalV, key string) *MetricsTraversalStep {
 	if tv.Error() != nil {
 		return NewMetricsTraversalStepFromError(tv.Error())
 	}
 
-	tv = tv.Dedup(ctx, "ID", "LastUpdateMetric.Start").Sort(ctx, common.SortAscending, "LastUpdateMetric.Start")
+	startField := key + ".Start"
+
+	tv = tv.Dedup(ctx, "ID", startField).Sort(ctx, common.SortAscending, startField)
+
 	if tv.Error() != nil {
 		return NewMetricsTraversalStepFromError(tv.Error())
 	}
@@ -57,18 +59,18 @@ nodeloop:
 			break nodeloop
 		}
 
-		m, _ := n.GetField("LastUpdateMetric")
+		m, _ := n.GetField(key)
 		if m == nil {
 			continue
 		}
 
-		lastMetric, ok := m.(*topology.InterfaceMetric)
+		lastmetric, ok := m.(common.Metric)
 		if !ok {
 			return NewMetricsTraversalStepFromError(errors.New("wrong interface metric type"))
 		}
 
-		if gslice == nil || (lastMetric.Start > gslice.Start && lastMetric.Last < gslice.Last) && it.Next() {
-			metrics[string(n.ID)] = append(metrics[string(n.ID)], lastMetric)
+		if gslice == nil || (lastmetric.GetStart() > gslice.Start && lastmetric.GetLast() < gslice.Last) && it.Next() {
+			metrics[string(n.ID)] = append(metrics[string(n.ID)], lastmetric)
 		}
 	}