sflow: fix metrics not having start set and protect aggregates

Author: safchain

gremlin/traversal/metrics.go

@@ -33,6 +33,7 @@ import (
 
 const (
 	defaultAggregatesSliceLength = int64(30000) // 30 seconds
+	aggregatesMaxSlices          = 10000
 )
 
 // MetricsTraversalExtension describes a new extension to enhance the topology
@@ -268,6 +269,10 @@ func (m *MetricsTraversalStep) Aggregates(ctx traversal.StepContext, s ...interf
 		steps++
 	}
 
+	if steps > aggregatesMaxSlices {
+		return NewMetricsTraversalStepFromError(fmt.Errorf("Aggregates available slices exceeded: %d/%d", steps, aggregatesMaxSlices))
+	}
+
 	aggregated := make([]common.Metric, steps, steps)
 	for _, metrics := range m.metrics {
 		aggregateMetrics(metrics, start, last, sliceLength, aggregated)

sflow/agent.go

@@ -109,7 +109,7 @@ func (sfa *Agent) feedFlowTable() {
 		}
 
 		// SFlow Counter Samples
-		var Countersamples []layers.SFlowCounterSample
+		var counterSamples []layers.SFlowCounterSample
 		var gen layers.SFlowGenericInterfaceCounters
 		var ovsdp layers.SFlowOVSDPCounters
 		var app layers.SFlowAppresourcesCounters
@@ -244,12 +244,9 @@ func (sfa *Agent) feedFlowTable() {
 
 				}
 			}
-			Countersamples = append(Countersamples, counter)
+			counterSamples = append(counterSamples, counter)
 		}
 
-		sfa.Graph.Lock()
-		tr := sfa.Graph.StartMetadataTransaction(sfa.Node)
-
 		currMetric := &SFMetric{
 			IfInOctets:         int64(gen.IfInOctets),
 			IfInUcastPkts:      int64(gen.IfInUcastPkts),
@@ -282,37 +279,36 @@ func (sfa *Agent) feedFlowTable() {
 			VlanBroadcastPkts:  int64(vlan.BroadcastPkts),
 			VlanDiscards:       int64(vlan.Discards),
 		}
-		now := int64(common.UnixMillis(time.Now()))
 
+		if currMetric.IsZero() {
+			continue
+		}
+
+		now := int64(common.UnixMillis(time.Now()))
 		currMetric.Last = now
 
-		var prevMetric, lastUpdateMetric, totalMetric *SFMetric
+		sfa.Graph.Lock()
 
+		var totalMetric *SFMetric
 		if metric, err := sfa.Node.GetField("SFlow.Metric"); err == nil {
-			prevMetric = metric.(*SFMetric)
-			lastUpdateMetric = currMetric
-			totalMetric = currMetric.Add(prevMetric).(*SFMetric)
+			prevMetric := metric.(*SFMetric)
+			currMetric.Start = prevMetric.Last
+
+			totalMetric = prevMetric.Add(currMetric).(*SFMetric)
+			totalMetric.Start = prevMetric.Start
 			totalMetric.Last = now
 		} else {
+			currMetric.Start = now
 			totalMetric = currMetric
 		}
 
-		// nothing changed since last update
-		if lastUpdateMetric != nil && !lastUpdateMetric.IsZero() {
-			lastUpdateMetric.Start = prevMetric.Last
-			lastUpdateMetric.Last = now
-		} else {
-			lastUpdateMetric = currMetric
-		}
-
 		sfl := &SFlow{
-			Counters:         Countersamples,
+			Counters:         counterSamples,
 			Metric:           totalMetric,
-			LastUpdateMetric: lastUpdateMetric,
+			LastUpdateMetric: currMetric,
 		}
 
-		tr.AddMetadata("SFlow", sfl)
-		tr.Commit()
+		sfa.Graph.AddMetadata(sfa.Node, "SFlow", sfl)
 		sfa.Graph.Unlock()
 	}
 }

tests/tests.go

@@ -513,6 +513,9 @@ func RunTest(t *testing.T, test *Test) {
 		t.Fatalf("Failed to setup captures: %s", err)
 	}
 
+	// wait a bit after the capture creation
+	time.Sleep(2 * time.Second)
+
 	retries := test.retries
 	if retries <= 0 {
 		retries = 30