issue#93: Sustaining request during specific seconds

Author: wmarta-edrans

cli/cli.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"os"
 	"os/signal"
 	"sort"
@@ -25,6 +26,7 @@ var (
 	url         string
 	concurrency uint
 	requests    uint
+	execTimeout uint
 	timeout     uint
 	regions     string
 	method      string
@@ -45,6 +47,7 @@ func main() {
 	flag.StringVar(&body, "b", "", "HTTP request body")
 	flag.UintVar(&concurrency, "c", 10, "number of concurrent requests")
 	flag.UintVar(&requests, "n", 1000, "number of total requests to make")
+	flag.UintVar(&execTimeout, "N", 0, "Maximum execution time in seconds")
 	flag.UintVar(&timeout, "t", 15, "request timeout in seconds")
 	flag.StringVar(&regions, "r", "us-east-1,eu-west-1,ap-northeast-1", "AWS regions to run in (comma separated, no spaces)")
 	flag.StringVar(&awsProfile, "p", "", "AWS named profile to use")
@@ -67,6 +70,7 @@ func main() {
 		URL:            url,
 		Concurrency:    concurrency,
 		TotalRequests:  requests,
+		ExecTimeout:    execTimeout,
 		RequestTimeout: time.Duration(timeout) * time.Second,
 		Regions:        strings.Split(regions, ","),
 		Method:         method,
@@ -120,6 +124,7 @@ func start(test *goad.Test, finalResult *queue.RegionsAggData, sigChan chan os.S
 		}
 	}()
 
+	startTime := time.Now()
 	firstTime := true
 outer:
 	for {
@@ -150,7 +155,12 @@ outer:
 			}
 
 			y = 0
-			percentDone := float64(totalReqs) / float64(result.TotalExpectedRequests)
+			var percentDone float64
+			if result.TotalExpectedRequests > 0 {
+				percentDone = float64(totalReqs) / float64(result.TotalExpectedRequests)
+			} else {
+				percentDone = math.Min(float64(time.Since(startTime).Seconds()) / float64(test.Config.ExecTimeout), 1.0)
+			}
 			drawProgressBar(percentDone, y)
 
 			termbox.Flush()

goad.go

@@ -25,6 +25,7 @@ type TestConfig struct {
 	URL            string
 	Concurrency    uint
 	TotalRequests  uint
+	ExecTimeout    uint
 	RequestTimeout time.Duration
 	Regions        []string
 	Method         string
@@ -53,7 +54,7 @@ var supportedRegions = []string{
 
 // Test type
 type Test struct {
-	config *TestConfig
+	Config *TestConfig
 	infra  *infrastructure.Infrastructure
 }
 
@@ -63,22 +64,22 @@ func NewTest(config *TestConfig) (*Test, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Test{config: config, infra: nil}, nil
+	return &Test{Config: config, infra: nil}, nil
 }
 
 // Start a test
 func (t *Test) Start() <-chan queue.RegionsAggData {
-	awsConfig := aws.NewConfig().WithRegion(t.config.Regions[0])
+	awsConfig := aws.NewConfig().WithRegion(t.Config.Regions[0])
 
-	if t.config.AwsProfile != "" {
-		creds := credentials.NewSharedCredentials("", t.config.AwsProfile)
+	if t.Config.AwsProfile != "" {
+		creds := credentials.NewSharedCredentials("", t.Config.AwsProfile)
 		if _, err := creds.Get(); err != nil {
 			log.Fatal(err)
 		}
 		awsConfig.WithCredentials(creds)
 	}
 
-	infra, err := infrastructure.New(t.config.Regions, awsConfig)
+	infra, err := infrastructure.New(t.Config.Regions, awsConfig)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -89,7 +90,7 @@ func (t *Test) Start() <-chan queue.RegionsAggData {
 	results := make(chan queue.RegionsAggData)
 
 	go func() {
-		for result := range queue.Aggregate(awsConfig, infra.QueueURL(), t.config.TotalRequests) {
+		for result := range queue.Aggregate(awsConfig, infra.QueueURL(), t.Config.TotalRequests) {
 			results <- result
 		}
 		close(results)
@@ -99,25 +100,28 @@ func (t *Test) Start() <-chan queue.RegionsAggData {
 }
 
 func (t *Test) invokeLambdas(awsConfig *aws.Config, sqsURL string) {
-	lambdas := numberOfLambdas(t.config.Concurrency, len(t.config.Regions))
+	lambdas := numberOfLambdas(t.Config.Concurrency, len(t.Config.Regions))
 
 	for i := 0; i < lambdas; i++ {
-		region := t.config.Regions[i%len(t.config.Regions)]
-		requests, requestsRemainder := divide(t.config.TotalRequests, lambdas)
-		concurrency, _ := divide(t.config.Concurrency, lambdas)
+		region := t.Config.Regions[i%len(t.Config.Regions)]
+		requests, requestsRemainder := divide(t.Config.TotalRequests, lambdas)
+		concurrency, _ := divide(t.Config.Concurrency, lambdas)
+		execTimeout := t.Config.ExecTimeout
 
 		if requestsRemainder > 0 && i == lambdas-1 {
 			requests += requestsRemainder
 		}
 
-		c := t.config
+		c := t.Config
 		args := []string{
 			"-u",
 			fmt.Sprintf("%s", c.URL),
 			"-c",
 			fmt.Sprintf("%s", strconv.Itoa(int(concurrency))),
 			"-n",
 			fmt.Sprintf("%s", strconv.Itoa(int(requests))),
+			"-N",
+			fmt.Sprintf("%s", strconv.Itoa(int(execTimeout))),
 			"-s",
 			fmt.Sprintf("%s", sqsURL),
 			"-q",
@@ -134,7 +138,7 @@ func (t *Test) invokeLambdas(awsConfig *aws.Config, sqsURL string) {
 			fmt.Sprintf("%s", c.Body),
 		}
 
-		for _, v := range t.config.Headers {
+		for _, v := range t.Config.Headers {
 			args = append(args, "-H", fmt.Sprintf("%s", v))
 		}
 
@@ -189,9 +193,12 @@ func (c TestConfig) check() error {
 	if c.Concurrency < 1 || c.Concurrency > concurrencyLimit {
 		return fmt.Errorf("Invalid concurrency (use 1 - %d)", concurrencyLimit)
 	}
-	if c.TotalRequests < 1 || c.TotalRequests > 2000000 {
+	if (c.TotalRequests < 1 && c.ExecTimeout <= 0) || c.TotalRequests > 2000000 {
 		return errors.New("Invalid total requests (use 1 - 2000000)")
 	}
+	if c.ExecTimeout > 300 {
+		return errors.New("Invalid maximum execution time in seconds (use 0 - 300)")
+	}
 	if c.RequestTimeout.Nanoseconds() < nano || c.RequestTimeout.Nanoseconds() > nano*100 {
 		return errors.New("Invalid timeout (1s - 100s)")
 	}

lambda/lambda.go

@@ -19,13 +19,16 @@ import (
 	"github.com/goadapp/goad/queue"
 )
 
+const lambdaTimeout = 295
+
 func main() {
 
 	var (
 		address          string
 		sqsurl           string
 		concurrencycount int
 		maxRequestCount  int
+		execTimeout      int
 		timeout          string
 		frequency        string
 		awsregion        string
@@ -46,10 +49,15 @@ func main() {
 
 	flag.IntVar(&concurrencycount, "c", 10, "number of concurrent requests")
 	flag.IntVar(&maxRequestCount, "n", 1000, "number of total requests to make")
+	flag.IntVar(&execTimeout, "N", 0, "Maximum execution time in seconds")
 
 	flag.Var(&requestHeaders, "H", "List of headers")
 	flag.Parse()
 
+	if execTimeout <= 0 || execTimeout > lambdaTimeout {
+		execTimeout = lambdaTimeout
+	}
+
 	clientTimeout, _ := time.ParseDuration(timeout)
 	fmt.Printf("Using a timeout of %s\n", clientTimeout)
 	reportingFrequency, _ := time.ParseDuration(frequency)
@@ -63,7 +71,7 @@ func main() {
 	client.Timeout = clientTimeout
 
 	fmt.Printf("Will spawn %d workers making %d requests to %s\n", concurrencycount, maxRequestCount, address)
-	runLoadTest(client, sqsurl, address, maxRequestCount, concurrencycount, awsregion, reportingFrequency, queueRegion, requestMethod, requestBody, requestHeaders)
+	runLoadTest(client, sqsurl, address, maxRequestCount, execTimeout, concurrencycount, awsregion, reportingFrequency, queueRegion, requestMethod, requestBody, requestHeaders)
 }
 
 type RequestResult struct {
@@ -80,7 +88,7 @@ type RequestResult struct {
 	State            string `json:"state"`
 }
 
-func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests int, concurrencycount int, awsregion string, reportingFrequency time.Duration, queueRegion string, requestMethod string, requestBody string, requestHeaders []string) {
+func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests int, execTimeout int, concurrencycount int, awsregion string, reportingFrequency time.Duration, queueRegion string, requestMethod string, requestBody string, requestHeaders []string) {
 	awsConfig := aws.NewConfig().WithRegion(queueRegion)
 	sqsAdaptor := queue.NewSQSAdaptor(awsConfig, sqsurl)
 	//sqsAdaptor := queue.NewDummyAdaptor(sqsurl)
@@ -102,10 +110,10 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 	fmt.Println(" done.\nWaiting for results…")
 
 	ticker := time.NewTicker(reportingFrequency)
-	quit := make(chan struct{})
+	quit := time.NewTimer(time.Duration(execTimeout) * time.Second)
 	quitting := false
 
-	for requestsSoFar < totalRequests && !quitting {
+	for (totalRequests == 0 || requestsSoFar < totalRequests) && !quitting {
 		i := 0
 
 		var timeToFirstTotal int64
@@ -120,7 +128,7 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 		var totalConnectionError int
 
 		resetStats := false
-		for requestsSoFar < totalRequests && !quitting && !resetStats {
+		for (totalRequests == 0 || requestsSoFar < totalRequests) && !quitting && !resetStats {
 			select {
 			case r := <-ch:
 				i++
@@ -164,12 +172,14 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 					statuses[statusStr]++
 				}
 				requestTimeTotal += r.Elapsed
+
 			case <-ticker.C:
 				if i == 0 {
 					continue
 				}
 				resetStats = true
-			case <-quit:
+
+			case <-quit.C:
 				ticker.Stop()
 				quitting = true
 			}
@@ -198,6 +208,7 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 			avgRequestTime = 0
 		}
 
+		finished := (totalRequests > 0 && requestsSoFar == totalRequests) || quitting
 		fatalError := ""
 		if (totalTimedOut + totalConnectionError) > i/2 {
 			fatalError = "Over 50% of requests failed, aborting"
@@ -216,6 +227,7 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 			fastest,
 			awsregion,
 			fatalError,
+			finished,
 		}
 		sqsAdaptor.SendResult(aggData)
 	}
@@ -225,7 +237,13 @@ func runLoadTest(client *http.Client, sqsurl string, url string, totalRequests i
 
 func fetch(loadTestStartTime time.Time, client *http.Client, address string, requestcount int, jobs <-chan struct{}, ch chan RequestResult, wg *sync.WaitGroup, awsregion string, requestMethod string, requestBody string, requestHeaders []string) {
 	defer wg.Done()
-	for _ = range jobs {
+	for {
+		if requestcount > 0 {
+			_, ok := <- jobs
+			if !ok {
+				break
+			}
+		}
 		start := time.Now()
 		req, err := http.NewRequest(requestMethod, address, bytes.NewBufferString(requestBody))
 		if err != nil {

queue/aggregation.go

@@ -21,6 +21,7 @@ type AggData struct {
 	Fastest              int64          `json:"fastest"`
 	Region               string         `json:"region"`
 	FatalError           string         `json:"fatal-error"`
+	Finished             bool           `json:"finished"`
 }
 
 // RegionsAggData type
@@ -31,12 +32,16 @@ type RegionsAggData struct {
 
 func (d *RegionsAggData) allRequestsReceived() bool {
 	var requests uint
+	var finishedCount int
 
 	for _, region := range d.Regions {
 		requests += uint(region.TotalReqs)
+		if region.Finished {
+			finishedCount += 1
+		}
 	}
 
-	return requests == d.TotalExpectedRequests
+	return requests == d.TotalExpectedRequests || finishedCount == len(d.Regions)
 }
 
 func addResult(data *AggData, result *AggData, isFinalSum bool) {
@@ -71,6 +76,9 @@ func addResult(data *AggData, result *AggData, isFinalSum bool) {
 	if result.Fastest > 0 && (data.Fastest == 0 || result.Fastest < data.Fastest) {
 		data.Fastest = result.Fastest
 	}
+	if result.Finished {
+		data.Finished = true
+	}
 }
 
 // SumRegionResults adds all the results together