Merge pull request #1632 from lebauce/fix-interface-metrics-race

traversal: fix race in interface metrics step

Author: safchain

cmd/allinone/allinone.go

@@ -129,7 +129,7 @@ var AllInOneCmd = &cobra.Command{
 
 		os.Setenv("SKYDIVE_ANALYZERS", fmt.Sprintf("%s:%d", addr, svcAddr.Port))
 		os.Setenv("SKYDIVE_LOGGING_FILE_PATH", logFile+"-agent"+extension)
-		
+
 		agentAttr := &os.ProcAttr{
 			Files: []*os.File{os.Stdin, os.Stdout, os.Stderr},
 			Env:   os.Environ(),

graffiti/graph/traversal/traversal.go

@@ -885,7 +885,9 @@ func (tv *GraphTraversalV) Sort(ctx StepContext, keys ...interface{}) *GraphTrav
 		sortBy = defaultSortBy
 	}
 
+	tv.GraphTraversal.RLock()
 	graph.SortNodes(tv.nodes, sortBy, sortOrder)
+	tv.GraphTraversal.RUnlock()
 
 	return tv
 }

gremlin/traversal/metrics.go

@@ -28,6 +28,7 @@ import (
 
 const (
 	defaultAggregatesSliceLength = int64(30000) // 30 seconds
+	aggregatesMaxSlices          = 10000
 )
 
 // MetricsTraversalExtension describes a new extension to enhance the topology
@@ -263,6 +264,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)

gremlin/traversal/topology.go

@@ -34,20 +34,19 @@ func InterfaceMetrics(ctx traversal.StepContext, tv *traversal.GraphTraversalV,
 	}
 
 	startField := key + ".Start"
-
 	tv = tv.Dedup(ctx, "ID", startField).Sort(ctx, common.SortAscending, startField)
 
 	if tv.Error() != nil {
 		return NewMetricsTraversalStepFromError(tv.Error())
 	}
 
+	tv.GraphTraversal.RLock()
+	defer tv.GraphTraversal.RUnlock()
+
 	metrics := make(map[string][]common.Metric)
 	it := ctx.PaginationRange.Iterator()
 	gslice := tv.GraphTraversal.Graph.GetContext().TimeSlice
 
-	tv.GraphTraversal.RLock()
-	defer tv.GraphTraversal.RUnlock()
-
 nodeloop:
 	for _, n := range tv.GetNodes() {
 		if it.Done() {

scripts/ci/run-tests-utils.sh

@@ -26,7 +26,7 @@ tests_run() {
         fi
 
         make test.functionals.batch \
-                GOFLAGS="$GOFLAGS" VERBOSE=true TAGS="$TAGS" GORACE="history_size=5" TIMEOUT=20m \
+                GOFLAGS="$GOFLAGS" VERBOSE=true TAGS="$TAGS" GORACE="history_size=7" TIMEOUT=20m \
                 WITH_HELM="$WITH_HELM" WITH_EBPF="$WITH_EBPF" WITH_K8S="$WITH_K8S" WITH_ISTIO="$WITH_ISTIO" \
                 ARGS="$ARGS" TEST_PATTERN="$TEST_PATTERN" 2>&1 | tee $LOGFILE
         RETCODE=$?