Add --max-iterations-per-second (#132)

<!--- Note to EXTERNAL Contributors -->
<!-- Thanks for opening a PR! 
If it is a significant code change, please **make sure there is an open
issue** for this.
We work best with you when we have accepted the idea first before you
code. -->

<!--- For ALL Contributors 👇 -->

## What was changed
<!-- Describe what has changed in this PR -->

Add RPS parameter for executor.

## Why?
<!-- Tell your future self why have you made these changes -->

To generate continuous but throttled load.

The new parameter allows to create a scenario where a large number of
iterations are started one-by-one at a constant rate (e.g. 5/s).

This is useful for creating a backlog of workflows (by turning off the
workers but keep creating new workflows).

## Checklist
<!--- add/delete as needed --->

1. Closes <!-- add issue number here -->

2. How was this tested:
<!--- Please describe how you tested your changes/how we can test them
-->

3. Any docs updates needed?
<!--- update README if applicable
      or point out where to update docs.temporal.io -->

Author: stephanos

cmd/run_scenario.go

@@ -52,6 +52,7 @@ type scenarioRunConfig struct {
 	iterations                    int
 	duration                      time.Duration
 	maxConcurrent                 int
+	maxIterationsPerSecond        float64
 	scenarioOptions               []string
 	timeout                       time.Duration
 	doNotRegisterSearchAttributes bool
@@ -75,6 +76,8 @@ func (r *scenarioRunConfig) addCLIFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&r.iterations, "iterations", 0, "Override default iterations for the scenario (cannot be provided with duration)")
 	fs.DurationVar(&r.duration, "duration", 0, "Override duration for the scenario (cannot be provided with iteration)."+
 		" This is the amount of time for which we will start new iterations of the scenario.")
+	fs.Float64Var(&r.maxIterationsPerSecond, "max-iterations-per-second", 0, "Override iterations per second rate limit for the scenario."+
+		" This is the maximum rate at which we will start new iterations of the scenario.")
 	fs.DurationVar(&r.timeout, "timeout", 0, "If set, the scenario will stop after this amount of"+
 		" time has elapsed. Any still-running iterations will be cancelled, and omes will exit nonzero.")
 	fs.IntVar(&r.maxConcurrent, "max-concurrent", 0, "Override max-concurrent for the scenario")
@@ -136,6 +139,7 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 			Iterations:                    r.iterations,
 			Duration:                      r.duration,
 			MaxConcurrent:                 r.maxConcurrent,
+			MaxIterationsPerSecond:        r.maxIterationsPerSecond,
 			Timeout:                       r.timeout,
 			DoNotRegisterSearchAttributes: r.doNotRegisterSearchAttributes,
 		},

cmd/run_scenario_with_worker.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"github.com/temporalio/omes/cmd/cmdoptions"
@@ -64,6 +65,7 @@ func (r *workerWithScenarioRunner) run(ctx context.Context) error {
 			iterations:                    r.iterations,
 			duration:                      r.duration,
 			maxConcurrent:                 r.maxConcurrent,
+			maxIterationsPerSecond:        r.maxIterationsPerSecond,
 			scenarioOptions:               r.scenarioOptions,
 			timeout:                       r.timeout,
 			doNotRegisterSearchAttributes: r.doNotRegisterSearchAttributes,

loadgen/generic_executor.go

@@ -57,6 +57,9 @@ func (g *GenericExecutor) newRun(info ScenarioInfo) (*genericRun, error) {
 	if run.config.Timeout == 0 {
 		run.config.Timeout = g.DefaultConfiguration.Timeout
 	}
+	if run.config.MaxIterationsPerSecond == 0 {
+		run.config.MaxIterationsPerSecond = g.DefaultConfiguration.MaxIterationsPerSecond
+	}
 	run.config.ApplyDefaults()
 	if run.config.Iterations > 0 && run.config.Duration > 0 {
 		return nil, fmt.Errorf("invalid scenario: iterations and duration are mutually exclusive")
@@ -113,9 +116,21 @@ func (g *genericRun) Run(ctx context.Context) error {
 		g.logger.Debugf("Will start iterations for %v", g.config.Duration)
 	}
 
+	var rateLimiter <-chan time.Time
+	if g.config.MaxIterationsPerSecond > 0 {
+		g.logger.Debugf("Will run at rate of %v iteration(s) per second", g.config.MaxIterationsPerSecond)
+		rateLimiter = time.Tick(time.Duration(float64(time.Second) / g.config.MaxIterationsPerSecond))
+	}
+
 	// Run all until we've gotten an error or reached iteration limit or timeout
 	for i := 0; runErr == nil && durationCtx.Err() == nil &&
 		(g.config.Iterations == 0 || i < g.config.Iterations); i++ {
+
+		// If there is a rate limit, enforce it
+		if rateLimiter != nil {
+			<-rateLimiter
+		}
+
 		// If there are already MaxConcurrent running, wait for one
 		if currentlyRunning >= g.config.MaxConcurrent {
 			waitOne(durationCtx)

loadgen/generic_executor_test.go

@@ -189,3 +189,22 @@ func TestRunIterationsWithoutTimeout(t *testing.T) {
 	require.NoError(t, err)
 	tracker.assertSeen(t, 5)
 }
+
+func TestRunIterationsWithRateLimit(t *testing.T) {
+	startTime := time.Now()
+	tracker := newIterationTracker()
+	err := execute(&GenericExecutor{
+		Execute: func(ctx context.Context, run *Run) error {
+			tracker.track(run.Iteration)
+			return nil
+		},
+		DefaultConfiguration: RunConfiguration{
+			Iterations:             4,
+			MaxConcurrent:          1,
+			MaxIterationsPerSecond: 4.0,
+		},
+	})
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, time.Since(startTime), time.Second)
+	tracker.assertSeen(t, 4)
+}

loadgen/scenario.go

@@ -128,6 +128,9 @@ type RunConfiguration struct {
 	// Maximum number of instances of the Execute method to run concurrently.
 	// Default is DefaultMaxConcurrent.
 	MaxConcurrent int
+	// MaxIterationsPerSecond is the maximum number of iterations to run per second.
+	// Default is zero, meaning unlimited.
+	MaxIterationsPerSecond float64
 	// Timeout is the maximum amount of time we'll wait for the scenario to finish running.
 	// If the timeout is hit any pending executions will be cancelled and the scenario will exit
 	// with an error. The default is unlimited.