Skip samples with any execution error (#33)

Signed-off-by: Raul Sevilla <[email protected]>

go.mod

@@ -43,6 +43,7 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/oauth2 v0.8.0 // indirect
+	golang.org/x/sync v0.3.0
 	golang.org/x/sys v0.9.0 // indirect
 	golang.org/x/term v0.8.0 // indirect
 	golang.org/x/text v0.9.0 // indirect

go.sum

@@ -1,5 +1,7 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
+github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/aws/aws-sdk-go v1.42.27/go.mod h1:OGr6lGMAKGlG9CVrYnWYDKIyb829c6EVBRjxqjmPepc=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -155,6 +157,8 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

pkg/runner/exec.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cloud-bulldozer/ingress-perf/pkg/config"
 	"github.com/cloud-bulldozer/ingress-perf/pkg/runner/tools"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -40,7 +41,6 @@ func runBenchmark(cfg config.Config, clusterMetadata ocpmetadata.ClusterMetadata
 	var timeouts, httpErrors int64
 	var benchmarkResult []tools.Result
 	var clientPods []corev1.Pod
-	var wg sync.WaitGroup
 	var ep string
 	var tool tools.Tool
 	r, err := orClientSet.RouteV1().Routes(benchmarkNs).Get(context.TODO(), fmt.Sprintf("%s-%s", serverName, cfg.Termination), metav1.GetOptions{})
@@ -80,15 +80,22 @@ func runBenchmark(cfg config.Config, clusterMetadata ocpmetadata.ClusterMetadata
 			Timestamp:       ts,
 			ClusterMetadata: clusterMetadata,
 		}
-		result.Config.Tuning = currentTuning // It's usefult to index the current tunning configuration in the all benchmark's documents
+		result.Config.Tuning = currentTuning // It's useful to index the current tuning patch in the all benchmark's documents
 		log.Infof("Running sample %d/%d: %v", i, cfg.Samples, cfg.Duration)
+		errGroup := errgroup.Group{}
 		for _, pod := range clientPods {
 			for i := 0; i < cfg.Procs; i++ {
-				wg.Add(1)
-				go exec(context.TODO(), &wg, tool, pod, &result)
+				func(p corev1.Pod) {
+					errGroup.Go(func() error {
+						return exec(context.TODO(), tool, p, &result)
+					})
+				}(pod)
 			}
 		}
-		wg.Wait()
+		if err = errGroup.Wait(); err != nil {
+			log.Error("Errors found during execution, skipping sample")
+			continue
+		}
 		genResultSummary(&result)
 		aggAvgRps += result.TotalAvgRps
 		aggAvgLatency += result.AvgLatency
@@ -114,8 +121,7 @@ func runBenchmark(cfg config.Config, clusterMetadata ocpmetadata.ClusterMetadata
 	return benchmarkResult, nil
 }
 
-func exec(ctx context.Context, wg *sync.WaitGroup, tool tools.Tool, pod corev1.Pod, result *tools.Result) {
-	defer wg.Done()
+func exec(ctx context.Context, tool tools.Tool, pod corev1.Pod, result *tools.Result) error {
 	var stdout, stderr bytes.Buffer
 	req := clientSet.CoreV1().RESTClient().Post().
 		Resource("pods").
@@ -133,27 +139,27 @@ func exec(ctx context.Context, wg *sync.WaitGroup, tool tools.Tool, pod corev1.P
 	exec, err := remotecommand.NewSPDYExecutor(restConfig, "POST", req.URL())
 	if err != nil {
 		log.Error(err.Error())
-		return
+		return err
 	}
 	err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
 		Stdout: &stdout,
 		Stderr: &stderr,
 	})
 	if err != nil {
-		log.Errorf("Exec failed, skipping: %v", err.Error())
-		return
+		log.Errorf("Exec failed in pod %s: %v", pod.Name, err.Error())
+		return err
 	}
 	podResult, err := tool.ParseResult(stdout.String(), stderr.String())
 	if err != nil {
-		log.Errorf("Result parsing failed, skipping: %v", err.Error())
-		return
+		log.Errorf("Result parsing failed: %v", err.Error())
+		return err
 	}
 	podResult.Name = pod.Name
 	podResult.Node = pod.Spec.NodeName
 	node, err := clientSet.CoreV1().Nodes().Get(context.TODO(), podResult.Node, metav1.GetOptions{})
 	if err != nil {
 		log.Errorf("Couldn't fetch node: %v", err.Error())
-		return
+		return err
 	}
 	if d, ok := node.Labels["node.kubernetes.io/instance-type"]; ok {
 		podResult.InstanceType = d
@@ -162,6 +168,7 @@ func exec(ctx context.Context, wg *sync.WaitGroup, tool tools.Tool, pod corev1.P
 	result.Pods = append(result.Pods, podResult)
 	lock.Unlock()
 	log.Debugf("%s: avgRps: %.0f avgLatency: %.0f ms", podResult.Name, podResult.AvgRps, podResult.AvgLatency/1000)
+	return nil
 }
 
 func genResultSummary(result *tools.Result) {

pkg/runner/runner.go

@@ -87,7 +87,7 @@ func Start(uuid, baseUUID, baseIndex string, tolerancy int, indexer *indexers.In
 	}
 	for i, cfg := range config.Cfg {
 		cfg.UUID = uuid
-		log.Infof("Running test %d/%d ", i+1, len(config.Cfg))
+		log.Infof("Running test %d/%d", i+1, len(config.Cfg))
 		log.Infof("Tool:%s termination:%v servers:%d concurrency:%d procs:%d connections:%d duration:%v",
 			cfg.Tool,
 			cfg.Termination,