Rewrite metrics calculation and retrieval

This commit moves most of the aggregation logic into the cli leaving the
goad runners with a cleaner implementation of the metrics to transmit to
the cli.

We introduce a runner id that allows tracking of the individual
lambdas being executed on aws and later aggregation of results for
different aws regions.

The infrastructure package has been restructured to move more of the aws
specific functionality into the corresponding packages.

A VSCode debug configuration has been added to version control.

Closes goadapp#143

Author: cwaltken-edrans

.vscode/launch.json

@@ -0,0 +1,19 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+
+        {
+            "name": "Launch Cli",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "remotePath": "",
+            "port": 2345,
+            "host": "127.0.0.1",
+            "program": "${workspaceRoot}",
+            "env": {},
+            "args": [],
+            "showLog": true
+        }
+    ]
+}

api/api.go

@@ -0,0 +1,22 @@
+package api
+
+import "time"
+
+// RunnerResult defines the common API for goad runners to send data back to the
+// cli.
+type RunnerResult struct {
+	AveTimeForReq    int64          `json:"ave-time-for-req"`
+	AveTimeToFirst   int64          `json:"ave-time-to-first"`
+	Fastest          int64          `json:"fastest"`
+	FatalError       string         `json:"fatal-error"`
+	Finished         bool           `json:"finished"`
+	Region           string         `json:"region"`
+	RunnerID         int            `json:"runner-id"`
+	Slowest          int64          `json:"slowest"`
+	Statuses         map[string]int `json:"statuses"`
+	TimeDelta        time.Duration  `json:"time-delta"`
+	BytesRead        int            `json:"bytes-read"`
+	ConnectionErrors int            `json:"connection-errors"`
+	RequestCount     int            `json:"request-count"`
+	TimedOut         int            `json:"timed-out"`
+}

cli/cli.go

@@ -10,7 +10,6 @@ import (
 	"os"
 	"os/signal"
 	"reflect"
-	"sort"
 	"strconv"
 	"strings"
 	"syscall"
@@ -21,7 +20,7 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"github.com/goadapp/goad/goad"
-	"github.com/goadapp/goad/queue"
+	"github.com/goadapp/goad/result"
 	"github.com/goadapp/goad/version"
 	"github.com/nsf/termbox-go"
 )
@@ -82,17 +81,14 @@ func Run() {
 	config := aggregateConfiguration()
 	test := createGoadTest(config)
 
-	var finalResult queue.RegionsAggData
-	defer printSummary(&finalResult)
-
-	if config.Output != "" {
-		defer saveJSONSummary(*outputFile, &finalResult)
-	}
-
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) // but interrupts from kbd are blocked by termbox
 
-	start(test, &finalResult, sigChan)
+	result := start(test, sigChan)
+	defer printSummary(result)
+	if config.Output != "" {
+		defer saveJSONSummary(*outputFile, result)
+	}
 }
 
 func writeIniFile() {
@@ -298,7 +294,8 @@ func createGoadTest(config *goad.TestConfig) *goad.Test {
 	return test
 }
 
-func start(test *goad.Test, finalResult *queue.RegionsAggData, sigChan chan os.Signal) {
+func start(test *goad.Test, sigChan chan os.Signal) result.LambdaResults {
+	var currentResult result.LambdaResults
 	resultChan, teardown := test.Start()
 	defer teardown()
 
@@ -337,43 +334,37 @@ outer:
 			if !ok {
 				break outer
 			}
-
+			currentResult = *result
 			if firstTime {
 				clearLogo()
 				firstTime = false
 			}
 
-			// sort so that regions always appear in the same order
-			var regions []string
-			for key := range result.Regions {
-				regions = append(regions, key)
-			}
-			sort.Strings(regions)
 			y := 3
 			totalReqs := 0
-			for _, region := range regions {
-				data := result.Regions[region]
-				totalReqs += data.TotalReqs
-				y = renderRegion(data, y)
+			regionsData := currentResult.RegionsData()
+			for _, region := range currentResult.Regions() {
+				totalReqs += regionsData[region].TotalReqs
+				y = renderRegion(regionsData[region], y)
 				y++
 			}
 
 			y = 0
 			var percentDone float64
-			if result.TotalExpectedRequests > 0 {
-				percentDone = float64(totalReqs) / float64(result.TotalExpectedRequests)
+			if test.Config.Requests > 0 {
+				percentDone = float64(totalReqs) / float64(test.Config.Requests)
 			} else {
 				percentDone = math.Min(float64(time.Since(startTime).Seconds())/float64(test.Config.Timelimit), 1.0)
 			}
 			drawProgressBar(percentDone, y)
 
 			termbox.Flush()
-			finalResult.Regions = result.Regions
 
 		case <-sigChan:
 			break outer
 		}
 	}
+	return currentResult
 }
 
 func renderLogo() {
@@ -400,7 +391,7 @@ func clearLogo() {
 }
 
 // renderRegion returns the y for the next empty line
-func renderRegion(data queue.AggData, y int) int {
+func renderRegion(data result.AggData, y int) int {
 	x := 0
 	renderString(x, y, "Region: ", termbox.ColorWhite, termbox.ColorBlue)
 	x += 8
@@ -417,14 +408,14 @@ func renderRegion(data queue.AggData, y int) int {
 	headingStr = "   Slowest    Fastest   Timeouts  TotErrors"
 	renderString(x, y, headingStr, coldef|termbox.AttrBold, coldef)
 	y++
-	resultStr = fmt.Sprintf("  %7.3fs   %7.3fs %10d %10d", float64(data.Slowest)/nano, float64(data.Fastest)/nano, data.TotalTimedOut, totErrors(&data))
+	resultStr = fmt.Sprintf("  %7.3fs   %7.3fs %10d %10d", float64(data.Slowest)/nano, float64(data.Fastest)/nano, data.TotalTimedOut, totErrors(data))
 	renderString(x, y, resultStr, coldef, coldef)
 	y++
 
 	return y
 }
 
-func totErrors(data *queue.AggData) int {
+func totErrors(data result.AggData) int {
 	var okReqs int
 	for statusStr, value := range data.Statuses {
 		status, _ := strconv.Atoi(statusStr)
@@ -463,28 +454,29 @@ func boldPrintln(msg string) {
 	fmt.Printf("\033[1m%s\033[0m\n", msg)
 }
 
-func printData(data *queue.AggData) {
+func printData(data result.AggData) {
 	boldPrintln("   TotReqs   TotBytes    AvgTime    AvgReq/s  (post)unzip")
 	fmt.Printf("%10d %10s   %7.3fs  %10.2f %10s/s\n", data.TotalReqs, humanize.Bytes(uint64(data.TotBytesRead)), float64(data.AveTimeForReq)/nano, data.AveReqPerSec, humanize.Bytes(uint64(data.AveKBytesPerSec)))
 	boldPrintln("   Slowest    Fastest   Timeouts  TotErrors")
 	fmt.Printf("  %7.3fs   %7.3fs %10d %10d", float64(data.Slowest)/nano, float64(data.Fastest)/nano, data.TotalTimedOut, totErrors(data))
 	fmt.Println("")
 }
 
-func printSummary(result *queue.RegionsAggData) {
-	if len(result.Regions) == 0 {
+func printSummary(results result.LambdaResults) {
+	if len(results.Regions()) == 0 {
 		boldPrintln("No results received")
 		return
 	}
 	boldPrintln("Regional results")
 	fmt.Println("")
 
-	for region, data := range result.Regions {
+	regionsData := results.RegionsData()
+	for _, region := range results.Regions() {
 		fmt.Println("Region: " + region)
-		printData(&data)
+		printData(regionsData[region])
 	}
 
-	overall := queue.SumRegionResults(result)
+	overall := results.SumAllLambdas()
 
 	fmt.Println("")
 	boldPrintln("Overall")
@@ -498,20 +490,16 @@ func printSummary(result *queue.RegionsAggData) {
 	fmt.Println("")
 }
 
-func saveJSONSummary(path string, result *queue.RegionsAggData) {
-	if len(result.Regions) == 0 {
+func saveJSONSummary(path string, results result.LambdaResults) {
+	if len(results.Regions()) == 0 {
 		return
 	}
-	results := make(map[string]queue.AggData)
-
-	for region, data := range result.Regions {
-		results[region] = data
-	}
+	dataForRegions := results.RegionsData()
 
-	overall := queue.SumRegionResults(result)
+	overall := results.SumAllLambdas()
 
-	results["overall"] = *overall
-	b, err := json.MarshalIndent(results, "", "  ")
+	dataForRegions["overall"] = overall
+	b, err := json.MarshalIndent(dataForRegions, "", "  ")
 	if err != nil {
 		fmt.Println(err)
 		return

goad/goad.go

@@ -12,7 +12,7 @@ import (
 	"github.com/goadapp/goad/infrastructure"
 	"github.com/goadapp/goad/infrastructure/aws"
 	"github.com/goadapp/goad/infrastructure/docker"
-	"github.com/goadapp/goad/queue"
+	"github.com/goadapp/goad/result"
 )
 
 // TestConfig type
@@ -57,9 +57,10 @@ var supportedRegions = []string{
 
 // Test type
 type Test struct {
-	Config  *TestConfig
-	infra   infrastructure.Infrastructure
-	lambdas int
+	Config    *TestConfig
+	infra     infrastructure.Infrastructure
+	lambdas   int
+	currentID int
 }
 
 // NewTest returns a configured Test
@@ -72,7 +73,7 @@ func NewTest(config *TestConfig) (*Test, error) {
 }
 
 // Start a test
-func (t *Test) Start() (<-chan queue.RegionsAggData, func()) {
+func (t *Test) Start() (<-chan *result.LambdaResults, func()) {
 	awsConfig := aws.NewConfig().WithRegion(t.Config.Regions[0])
 
 	if t.Config.RunDocker {
@@ -87,10 +88,10 @@ func (t *Test) Start() (<-chan queue.RegionsAggData, func()) {
 	t.lambdas = numberOfLambdas(t.Config.Concurrency, len(t.Config.Regions))
 	t.invokeLambdas()
 
-	results := make(chan queue.RegionsAggData)
+	results := make(chan *result.LambdaResults)
 
 	go func() {
-		for result := range queue.Aggregate(awsConfig, t.infra.GetQueueURL(), t.Config.Requests, t.lambdas) {
+		for result := range result.Aggregate(awsConfig, t.infra.GetQueueURL(), t.Config.Requests, t.lambdas) {
 			results <- result
 		}
 		close(results)
@@ -121,8 +122,10 @@ func (t *Test) invokeLambdas() {
 			fmt.Sprintf("--frequency=%s", reportingFrequency(t.lambdas).String()),
 			fmt.Sprintf("--aws-region=%s", region),
 			fmt.Sprintf("--method=%s", c.Method),
+			fmt.Sprintf("--runner-id=%d", t.currentID),
 			fmt.Sprintf("--body=%s", c.Body),
 		}
+		t.currentID++
 		for _, v := range t.Config.Headers {
 			args = append(args, fmt.Sprintf("--header=%s", v))
 		}

goad/util/util.go

@@ -0,0 +1,13 @@
+package util
+
+func RemoveDuplicates(strs []string) []string {
+	strsMap := make(map[string]bool)
+	for _, str := range strs {
+		strsMap[str] = true
+	}
+	returnStrs := make([]string, 0)
+	for str := range strsMap {
+		returnStrs = append(returnStrs, str)
+	}
+	return returnStrs
+}