Merge pull request goadapp#169 from cwaltken-edrans/improve-messaging

Improve SQS messaging and overall infrastructure code

Author: zeph

Gopkg.lock

@@ -20,6 +20,7 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"github.com/goadapp/goad/goad"
+	"github.com/goadapp/goad/goad/types"
 	"github.com/goadapp/goad/result"
 	"github.com/goadapp/goad/version"
 	"github.com/nsf/termbox-go"
@@ -79,12 +80,13 @@ func Run() {
 	app.VersionFlag.Short('V')
 
 	config := aggregateConfiguration()
-	test := createGoadTest(config)
+	err := config.Check()
+	goad.HandleErr(err)
 
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) // but interrupts from kbd are blocked by termbox
 
-	result := start(test, sigChan)
+	result := start(config, sigChan)
 	defer printSummary(result)
 	if config.Output != "" {
 		defer saveJSONSummary(*outputFile, result)
@@ -102,15 +104,15 @@ func writeConfigStream(writer io.Writer) {
 	stream.WriteTo(writer)
 }
 
-func aggregateConfiguration() *goad.TestConfig {
+func aggregateConfiguration() *types.TestConfig {
 	config := parseSettings()
 	applyDefaultsFromConfig(config)
 	config = parseCommandline()
 	applyExtendedConfiguration(config)
 	return config
 }
 
-func applyDefaultsFromConfig(config *goad.TestConfig) {
+func applyDefaultsFromConfig(config *types.TestConfig) {
 	applyDefaultIfNotZero(bodyFlag, config.Body)
 	applyDefaultIfNotZero(concurrencyFlag, prepareInt(config.Concurrency))
 	applyDefaultIfNotZero(headersFlag, config.Headers)
@@ -191,8 +193,8 @@ func loadIni() *ini.File {
 	return cfg
 }
 
-func parseSettings() *goad.TestConfig {
-	config := &goad.TestConfig{}
+func parseSettings() *types.TestConfig {
+	config := &types.TestConfig{}
 	cfg := loadIni()
 	if cfg == nil {
 		return config
@@ -219,7 +221,7 @@ func parseSettings() *goad.TestConfig {
 	return config
 }
 
-func applyExtendedConfiguration(config *goad.TestConfig) {
+func applyExtendedConfiguration(config *types.TestConfig) {
 	cfg := loadIni()
 	if cfg == nil {
 		return
@@ -229,11 +231,7 @@ func applyExtendedConfiguration(config *goad.TestConfig) {
 	if err != nil {
 		return
 	}
-	task := goad.CustomTask{
-		// Name:       taskSection.Name(),
-		RunnerPath: runnerPathKey.String(),
-	}
-	config.Task = task
+	config.RunnerPath = runnerPathKey.String()
 }
 
 func foldHeaders(hash map[string]string) []string {
@@ -244,7 +242,7 @@ func foldHeaders(hash map[string]string) []string {
 	return headersList
 }
 
-func parseCommandline() *goad.TestConfig {
+func parseCommandline() *types.TestConfig {
 	args := os.Args[1:]
 
 	kingpin.MustParse(app.Parse(args))
@@ -262,7 +260,7 @@ func parseCommandline() *goad.TestConfig {
 
 	regionsArray := parseRegionsForBackwardsCompatibility(*regions)
 
-	config := &goad.TestConfig{}
+	config := &types.TestConfig{}
 	config.URL = *url
 	config.Concurrency = *concurrency
 	config.Requests = *requests
@@ -285,22 +283,13 @@ func parseRegionsForBackwardsCompatibility(regions []string) []string {
 	return parsedRegions
 }
 
-func createGoadTest(config *goad.TestConfig) *goad.Test {
-	test, err := goad.NewTest(config)
-	if err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
-	return test
-}
-
-func start(test *goad.Test, sigChan chan os.Signal) result.LambdaResults {
+func start(test *types.TestConfig, sigChan chan os.Signal) result.LambdaResults {
 	var currentResult result.LambdaResults
-	resultChan, teardown := test.Start()
+	resultChan, teardown := goad.Start(test)
 	defer teardown()
 
 	platform := "AWS"
-	if test.Config.RunDocker {
+	if test.RunDocker {
 		platform = "Docker"
 	}
 	launchingOn := fmt.Sprintf("Launching on %s... (be patient)", platform)
@@ -351,10 +340,10 @@ outer:
 
 			y = 0
 			var percentDone float64
-			if test.Config.Requests > 0 {
-				percentDone = float64(totalReqs) / float64(test.Config.Requests)
+			if test.Requests > 0 {
+				percentDone = float64(totalReqs) / float64(test.Requests)
 			} else {
-				percentDone = math.Min(float64(time.Since(startTime).Seconds())/float64(test.Config.Timelimit), 1.0)
+				percentDone = math.Min(float64(time.Since(startTime).Seconds())/float64(test.Timelimit), 1.0)
 			}
 			drawProgressBar(percentDone, y)
 

cli/cli.go

@@ -7,15 +7,14 @@ import (
 	"sort"
 	"testing"
 
-	"github.com/goadapp/goad/goad"
+	"github.com/goadapp/goad/goad/types"
 	"github.com/stretchr/testify/assert"
 )
 
 const testDataFile = "testdata/test-config.ini"
 
 var expectedRegions = []string{"us-east-1", "eu-west-1"}
 var expectedHeader = []string{"cache-control: no-cache", "auth-token: YOUR-SECRET-AUTH-TOKEN", "base64-header: dGV4dG8gZGUgcHJ1ZWJhIA=="}
-var expectedTask = goad.CustomTask{RunnerPath: "default-runner"}
 
 func TestLoadStandardConfig(t *testing.T) {
 	iniFile = testDataFile
@@ -24,7 +23,7 @@ func TestLoadStandardConfig(t *testing.T) {
 	assertConfigContent(config, t)
 }
 
-func assertConfigContent(config *goad.TestConfig, t *testing.T) {
+func assertConfigContent(config *types.TestConfig, t *testing.T) {
 	assert := assert.New(t)
 	assert.Equal("http://file-config.com/", config.URL, "Should load the URL")
 	assert.Equal("GET", config.Method, "Should load the request method")
@@ -38,7 +37,7 @@ func assertConfigContent(config *goad.TestConfig, t *testing.T) {
 	sort.Strings(expectedHeader)
 	sort.Strings(config.Headers)
 	assert.Equal(expectedHeader, config.Headers, "Should load the output file")
-	assert.Equal(expectedTask, config.Task, "Should load tasks configuration")
+	assert.Equal("default-runner", config.RunnerPath, "Should load runner path configuration")
 }
 
 func TestSaveConfig(t *testing.T) {

cli/ini_parser_test.go

@@ -1,97 +1,31 @@
 package goad
 
 import (
-	"errors"
-	"fmt"
-	"math"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
+	"github.com/goadapp/goad/goad/types"
 	"github.com/goadapp/goad/infrastructure"
 	"github.com/goadapp/goad/infrastructure/aws"
 	"github.com/goadapp/goad/infrastructure/docker"
 	"github.com/goadapp/goad/result"
 )
 
-// TestConfig type
-type TestConfig struct {
-	URL         string
-	Concurrency int
-	Requests    int
-	Timelimit   int
-	Timeout     int
-	Regions     []string
-	Method      string
-	Body        string
-	Headers     []string
-	Output      string
-	Settings    string
-	RunDocker   bool
-	Task        CustomTask
-}
-
-type CustomTask struct {
-	RunnerPath string
-}
-
-const (
-	MAX_REQUEST_COUNT = math.MaxInt32
-	nano              = 1000000000
-)
-
-var supportedRegions = []string{
-	"us-east-1",      // N. Virginia
-	"us-east-2",      // Ohio
-	"us-west-1",      // N.California
-	"us-west-2",      // Oregon
-	"eu-west-1",      // Ireland
-	"eu-central-1",   // Frankfurt
-	"ap-northeast-1", // Sydney
-	"ap-northeast-2", // Seoul
-	"ap-southeast-1", // Singapore
-	"ap-southeast-2", // Tokio
-	"sa-east-1",      // Sao Paulo
-}
-
-// Test type
-type Test struct {
-	Config    *TestConfig
-	infra     infrastructure.Infrastructure
-	lambdas   int
-	currentID int
-}
-
-// NewTest returns a configured Test
-func NewTest(config *TestConfig) (*Test, error) {
-	err := config.check()
-	if err != nil {
-		return nil, err
-	}
-	return &Test{Config: config, infra: nil}, nil
-}
-
 // Start a test
-func (t *Test) Start() (<-chan *result.LambdaResults, func()) {
-	awsConfig := aws.NewConfig().WithRegion(t.Config.Regions[0])
+func Start(t *types.TestConfig) (<-chan *result.LambdaResults, func()) {
 
-	if t.Config.RunDocker {
-		t.infra = dockerinfra.NewDockerInfrastructure()
+	var infra infrastructure.Infrastructure
+	if t.RunDocker {
+		infra = dockerinfra.New(t)
 	} else {
-		t.infra = awsinfra.New(t.Config.Regions, awsConfig)
+		infra = awsinfra.New(t)
 	}
-	teardown, err := t.infra.Setup(infrastructure.Settings{
-		RunnerPath: t.Config.Task.RunnerPath,
-	})
-	handleErr(err)
-	t.lambdas = numberOfLambdas(t.Config.Concurrency, len(t.Config.Regions))
-	t.invokeLambdas()
+	teardown, err := infra.Setup()
+	HandleErr(err)
+	t.Lambdas = numberOfLambdas(t.Concurrency, len(t.Regions))
+	infrastructure.InvokeLambdas(infra)
 
 	results := make(chan *result.LambdaResults)
 
 	go func() {
-		for result := range result.Aggregate(awsConfig, t.infra.GetQueueURL(), t.Config.Requests, t.lambdas) {
+		for result := range infrastructure.Aggregate(infra) {
 			results <- result
 		}
 		close(results)
@@ -100,47 +34,7 @@ func (t *Test) Start() (<-chan *result.LambdaResults, func()) {
 	return results, teardown
 }
 
-func (t *Test) invokeLambdas() {
-	for i := 0; i < t.lambdas; i++ {
-		region := t.Config.Regions[i%len(t.Config.Regions)]
-		requests, requestsRemainder := divide(t.Config.Requests, t.lambdas)
-		concurrency, _ := divide(t.Config.Concurrency, t.lambdas)
-		execTimeout := t.Config.Timelimit
-
-		if requestsRemainder > 0 && i == t.lambdas-1 {
-			requests += requestsRemainder
-		}
-
-		c := t.Config
-		args := []string{
-			fmt.Sprintf("--concurrency=%s", strconv.Itoa(int(concurrency))),
-			fmt.Sprintf("--requests=%s", strconv.Itoa(int(requests))),
-			fmt.Sprintf("--execution-time=%s", strconv.Itoa(int(execTimeout))),
-			fmt.Sprintf("--sqsurl=%s", t.infra.GetQueueURL()),
-			fmt.Sprintf("--queue-region=%s", c.Regions[0]),
-			fmt.Sprintf("--client-timeout=%s", time.Duration(c.Timeout)*time.Second),
-			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))
-		}
-		args = append(args, fmt.Sprintf("%s", c.URL))
-
-		invokeargs := infrastructure.InvokeArgs{
-			File: "./goad-lambda",
-			Args: args,
-		}
-
-		go t.infra.Run(invokeargs)
-	}
-}
-
-func handleErr(err error) {
+func HandleErr(err error) {
 	if err != nil {
 		panic(err)
 	}
@@ -162,45 +56,3 @@ func numberOfLambdas(concurrency int, numRegions int) int {
 	}
 	return int(concurrency-1)/10 + 1
 }
-
-func divide(dividend int, divisor int) (quotient, remainder int) {
-	return dividend / divisor, dividend % divisor
-}
-
-func reportingFrequency(numberOfLambdas int) time.Duration {
-	return time.Duration((math.Log2(float64(numberOfLambdas)) + 1)) * time.Second
-}
-
-func (c TestConfig) check() error {
-	concurrencyLimit := 25000 * len(c.Regions)
-	if c.Concurrency < 1 || c.Concurrency > concurrencyLimit {
-		return fmt.Errorf("Invalid concurrency (use 1 - %d)", concurrencyLimit)
-	}
-	if (c.Requests < 1 && c.Timelimit <= 0) || c.Requests > MAX_REQUEST_COUNT {
-		return errors.New(fmt.Sprintf("Invalid total requests (use 1 - %d)", MAX_REQUEST_COUNT))
-	}
-	if c.Timelimit > 3600 {
-		return errors.New("Invalid maximum execution time in seconds (use 0 - 3600)")
-	}
-	if c.Timeout < 1 || c.Timeout > 100 {
-		return errors.New("Invalid timeout (1s - 100s)")
-	}
-	for _, region := range c.Regions {
-		supportedRegionFound := false
-		for _, supported := range supportedRegions {
-			if region == supported {
-				supportedRegionFound = true
-			}
-		}
-		if !supportedRegionFound {
-			return fmt.Errorf("Unsupported region: %s. Supported regions are: %s.", region, strings.Join(supportedRegions, ", "))
-		}
-	}
-	for _, v := range c.Headers {
-		header := strings.Split(v, ":")
-		if len(header) < 2 {
-			return fmt.Errorf("Header %s not valid. Make sure your header is of the form \"Header: value\"", v)
-		}
-	}
-	return nil
-}