New config framework for the different schedulers

README.md

@@ -197,7 +197,6 @@ Configuration mechanisms do have an order of precedence. As presented, options a
 As shown above, there are several ways to configure the number of simultaneous virtual users k6 will launch. There are also different ways to specify how long those virtual users will be running. For simple tests you can:
 - Set the test duration by the `--duration`/`-d` CLI flag (or the `K6_DURATION` environment variable and the `duration` script/JSON option). For ease of use, `duration` is specified with human readable values like `1h30m10s` - `k6 run --duration 30s script.js`, `k6 cloud -d 15m10s script.js`, `export K6_DURATION=1h`, etc. If set to `0`, k6 wouldn't stop executing the script unless the user manually stops it.
 - Set the total number of script iterations with the `--iterations`/`-i` CLI flag (or the `K6_ITERATIONS` environment variable and the `iterations` script/JSON option). k6 will stop executing the script whenever the **total** number of iterations (i.e. the number of iterations across all VUs) reaches the specified number. So if you have `k6 run --iterations 10 --vus 10 script.js`, then each VU would make only a single iteration.
-- Set both the test duration and the total number of script iterations. In that case, k6 would stop the script execution whenever either one of the above conditions is reached first.
 
 For more complex cases, you can specify execution stages. They are a combination of `duration,target-VUs` pairs. These pairs instruct k6 to linearly ramp up, ramp down, or stay at the number of VUs specified for the period specified. Execution stages can be set via the `stages` script/JSON option as an array of `{ duration: ..., target: ... }` pairs, or with the `--stage`/`-s` CLI flags and the `K6_STAGE` environment variable via the `duration:target,duration:target...` syntax.
 

cmd/config.go

@@ -27,10 +27,13 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	"github.com/loadimpact/k6/lib"
+	"github.com/loadimpact/k6/lib/scheduler"
 	"github.com/loadimpact/k6/stats/cloud"
 	"github.com/loadimpact/k6/stats/influxdb"
 	"github.com/loadimpact/k6/stats/kafka"
+	"github.com/pkg/errors"
 	"github.com/shibukawa/configdir"
+	log "github.com/sirupsen/logrus"
 	"github.com/spf13/afero"
 	"github.com/spf13/pflag"
 	null "gopkg.in/guregu/null.v3"
@@ -171,6 +174,71 @@ func readEnvConfig() (conf Config, err error) {
 	return conf, nil
 }
 
+// This checks for conflicting options and turns any shortcut options (i.e. duration, iterations,
+// stages) into the proper scheduler configuration
+func buildExecutionConfig(conf Config) (Config, error) {
+	result := conf
+	if conf.Duration.Valid {
+		if conf.Iterations.Valid {
+			//TODO: make this an error in the next version
+			log.Warnf("Specifying both duration and iterations is deprecated and won't be supported in the future k6 versions")
+		}
+
+		if conf.Stages != nil {
+			//TODO: make this an error in the next version
+			log.Warnf("Specifying both duration and stages is deprecated and won't be supported in the future k6 versions")
+		}
+
+		if conf.Execution != nil {
+			return result, errors.New("specifying both duration and execution is not supported")
+		}
+
+		ds := scheduler.NewConstantLoopingVUsConfig(lib.DefaultSchedulerName)
+		ds.VUs = conf.VUs
+		ds.Duration = conf.Duration
+		result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
+	} else if conf.Stages != nil {
+		if conf.Iterations.Valid {
+			//TODO: make this an error in the next version
+			log.Warnf("Specifying both iterations and stages is deprecated and won't be supported in the future k6 versions")
+		}
+
+		if conf.Execution != nil {
+			return conf, errors.New("specifying both stages and execution is not supported")
+		}
+
+		ds := scheduler.NewVariableLoopingVUsConfig(lib.DefaultSchedulerName)
+		ds.StartVUs = conf.VUs
+		for _, s := range conf.Stages {
+			if s.Duration.Valid {
+				ds.Stages = append(ds.Stages, scheduler.Stage{Duration: s.Duration, Target: s.Target})
+			}
+		}
+		result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
+	} else if conf.Iterations.Valid || conf.Execution == nil {
+		// Either shared iterations were explicitly specified via the shortcut option, or no execution
+		// parameters were specified in any way, which will run the default 1 iteration in 1 VU
+		if conf.Iterations.Valid && conf.Execution != nil {
+			return conf, errors.New("specifying both iterations and execution is not supported")
+		}
+
+		ds := scheduler.NewSharedIterationsConfig(lib.DefaultSchedulerName)
+		ds.VUs = conf.VUs
+		if conf.Iterations.Valid { // TODO: fix where the default iterations value is set... sigh...
+			ds.Iterations = conf.Iterations
+		}
+
+		result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
+	}
+
+	//TODO: validate the config; questions:
+	// - separately validate the duration, iterations and stages for better error messages?
+	// - or reuse the execution validation somehow, at the end? or something mixed?
+	// - here or in getConsolidatedConfig() or somewhere else?
+
+	return result, nil
+}
+
 // Assemble the final consolidated configuration from all of the different sources:
 // - start with the CLI-provided options to get shadowed (non-Valid) defaults in there
 // - add the global file config options
@@ -179,6 +247,7 @@ func readEnvConfig() (conf Config, err error) {
 // - merge the user-supplied CLI flags back in on top, to give them the greatest priority
 // - set some defaults if they weren't previously specified
 // TODO: add better validation, more explicit default values and improve consistency between formats
+// TODO: accumulate all errors and differentiate between the layers?
 func getConsolidatedConfig(fs afero.Fs, cliConf Config, runner lib.Runner) (conf Config, err error) {
 	cliConf.Collectors.InfluxDB = influxdb.NewConfig().Apply(cliConf.Collectors.InfluxDB)
 	cliConf.Collectors.Cloud = cloud.NewConfig().Apply(cliConf.Collectors.Cloud)
@@ -199,5 +268,5 @@ func getConsolidatedConfig(fs afero.Fs, cliConf Config, runner lib.Runner) (conf
 	}
 	conf = conf.Apply(envConf).Apply(cliConf)
 
-	return conf, nil
+	return buildExecutionConfig(conf)
 }

cmd/config_test.go

@@ -134,3 +134,8 @@ func TestConfigApply(t *testing.T) {
 		assert.Equal(t, []string{"influxdb", "json"}, conf.Out)
 	})
 }
+
+func TestBuildExecutionConfig(t *testing.T) {
+	//TODO: test the current config building and constructing of the execution plan, and the emitted warnings
+	//TODO: test the future full overwriting of the duration/iterations/stages/execution options
+}

cmd/run.go

@@ -37,6 +37,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/davecgh/go-spew/spew"
+
 	"github.com/loadimpact/k6/api"
 	"github.com/loadimpact/k6/core"
 	"github.com/loadimpact/k6/core/local"
@@ -61,7 +63,7 @@ const (
 	teardownTimeoutErrorCode    = 101
 	genericTimeoutErrorCode     = 102
 	genericEngineErrorCode      = 103
-	invalidOptionsErrorCode     = 104
+	//invalidOptionsErrorCode     = 104
 )
 
 var (
@@ -167,14 +169,16 @@ a commandline interface for interacting with it.`,
 		}
 
 		//TODO: move a bunch of the logic above to a config "constructor" and to the Validate() method
+		spew.Dump(conf.Execution)
 		if errList := conf.Validate(); len(errList) != 0 {
-			errMsg := []string{
-				fmt.Sprintf("There were %d errors with the script options:", len(errList)),
-			}
+
+			errMsg := []string{"There were problems with the specified script configuration:"}
 			for _, err := range errList {
 				errMsg = append(errMsg, fmt.Sprintf("\t- %s", err.Error()))
 			}
-			return ExitCode{errors.New(strings.Join(errMsg, "\n")), invalidOptionsErrorCode}
+			//TODO: re-enable exiting with validation errors
+			log.Warn(errors.New(strings.Join(errMsg, "\n")))
+			//return ExitCode{errors.New(strings.Join(errMsg, "\n")), invalidOptionsErrorCode}
 		}
 
 		// If summary trend stats are defined, update the UI to reflect them

lib/options.go

@@ -34,6 +34,11 @@ import (
 	"gopkg.in/guregu/null.v3"
 )
 
+// DefaultSchedulerName is used as the default key/ID of the scheduler config entries
+// that were created due to the use of the shortcut execution control options (i.e. duration+vus,
+// iterations+vus, or stages)
+const DefaultSchedulerName = "default"
+
 // DefaultSystemTagList includes all of the system tags emitted with metrics by default.
 // Other tags that are not enabled by default include: iter, vu, ocsp_status, ip
 var DefaultSystemTagList = []string{
@@ -189,7 +194,9 @@ type Options struct {
 
 	// Initial values for VUs, max VUs, duration cap, iteration cap, and stages.
 	// See the Runner or Executor interfaces for more information.
-	VUs        null.Int           `json:"vus" envconfig:"vus"`
+	VUs null.Int `json:"vus" envconfig:"vus"`
+
+	//TODO: deprecate this? or reuse it in the manual control "scheduler"?
 	VUsMax     null.Int           `json:"vusMax" envconfig:"vus_max"`
 	Duration   types.NullDuration `json:"duration" envconfig:"duration"`
 	Iterations null.Int           `json:"iterations" envconfig:"iterations"`
@@ -296,6 +303,23 @@ func (o Options) Apply(opts Options) Options {
 	if opts.VUsMax.Valid {
 		o.VUsMax = opts.VUsMax
 	}
+
+	// Specifying duration, iterations, stages, or execution in a "higher" config tier
+	// will overwrite all of the the previous execution settings (if any) from any
+	// "lower" config tiers
+	// Still, if more than one of those options is simultaneously specified in the same
+	// config tier, they will be preserved, so the validation after we've consolidated
+	// all of the options can return an error.
+	if opts.Duration.Valid || opts.Iterations.Valid || opts.Stages != nil || o.Execution != nil {
+		//TODO: uncomment this after we start using the new schedulers
+		/*
+			o.Duration = types.NewNullDuration(0, false)
+			o.Iterations = null.NewInt(0, false)
+			o.Stages = nil
+			o.Execution = nil
+		*/
+	}
+
 	if opts.Duration.Valid {
 		o.Duration = opts.Duration
 	}
@@ -310,12 +334,13 @@ func (o Options) Apply(opts Options) Options {
 			}
 		}
 	}
-
-	//TODO: handle o.Execution overwriting by plain vus/iterations/duration/stages options
-	if len(opts.Execution) > 0 {
+	// o.Execution can also be populated by the duration/iterations/stages config shortcuts, but
+	// that happens after the configuration from the different sources is consolidated. It can't
+	// happen here, because something like `K6_ITERATIONS=10 k6 run --vus 5 script.js` wont't
+	// work correctly at this level.
+	if opts.Execution != nil {
 		o.Execution = opts.Execution
 	}
-
 	if opts.SetupTimeout.Valid {
 		o.SetupTimeout = opts.SetupTimeout
 	}

lib/options_test.go

@@ -30,6 +30,7 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	"github.com/loadimpact/k6/lib/scheduler"
 	"github.com/loadimpact/k6/lib/types"
 	"github.com/loadimpact/k6/stats"
 	"github.com/stretchr/testify/assert"
@@ -86,6 +87,17 @@ func TestOptions(t *testing.T) {
 		assert.Equal(t, oneStage, opts.Apply(Options{Stages: oneStage}).Stages)
 		assert.Equal(t, oneStage, Options{}.Apply(opts).Apply(Options{Stages: oneStage}).Apply(Options{Stages: oneStage}).Stages)
 	})
+	t.Run("Execution", func(t *testing.T) {
+		sched := scheduler.NewConstantLoopingVUsConfig("test")
+		sched.VUs = null.IntFrom(123)
+		sched.Duration = types.NullDurationFrom(3 * time.Minute)
+		opts := Options{}.Apply(Options{Execution: scheduler.ConfigMap{"test": sched}})
+		cs, ok := opts.Execution["test"].(scheduler.ConstantLoopingVUsConfig)
+		assert.True(t, ok)
+		assert.Equal(t, int64(123), cs.VUs.Int64)
+		assert.Equal(t, "3m0s", cs.Duration.String())
+	})
+	//TODO: test that any execution option overwrites any other lower-level options
 	t.Run("RPS", func(t *testing.T) {
 		opts := Options{}.Apply(Options{RPS: null.IntFrom(12345)})
 		assert.True(t, opts.RPS.Valid)

lib/scheduler/base_config.go

@@ -21,15 +21,12 @@ type BaseConfig struct {
 	Interruptible    null.Bool          `json:"interruptible"`
 	IterationTimeout types.NullDuration `json:"iterationTimeout"`
 	Env              map[string]string  `json:"env"`
-	Exec             string             `json:"exec"` // function name, externally validated
+	Exec             null.String        `json:"exec"` // function name, externally validated
 	Percentage       float64            `json:"-"`    // 100, unless Split() was called
 
-	// Future extensions: tags, distribution, others?
+	//TODO: future extensions like tags, distribution, others?
 }
 
-// Make sure we implement the Config interface, even with the BaseConfig!
-var _ Config = &BaseConfig{}
-
 // NewBaseConfig returns a default base config with the default values
 func NewBaseConfig(name, configType string, interruptible bool) BaseConfig {
 	return BaseConfig{
@@ -56,6 +53,9 @@ func (bc BaseConfig) Validate() (errors []error) {
 			"percentage should be between %f and 100, but is %f", minPercentage, bc.Percentage,
 		))
 	}
+	if bc.Exec.Valid && bc.Exec.String == "" {
+		errors = append(errors, fmt.Errorf("exec value cannot be empty"))
+	}
 	// The actually reasonable checks:
 	if bc.StartTime.Valid && bc.StartTime.Duration < 0 {
 		errors = append(errors, fmt.Errorf("scheduler start time should be positive"))
@@ -81,15 +81,3 @@ func (bc BaseConfig) CopyWithPercentage(percentage float64) *BaseConfig {
 	c.Percentage = percentage
 	return &c
 }
-
-// Split splits the BaseConfig with the accurate percentages
-func (bc BaseConfig) Split(percentages []float64) ([]Config, error) {
-	if err := checkPercentagesSum(percentages); err != nil {
-		return nil, err
-	}
-	configs := make([]Config, len(percentages))
-	for i, p := range percentages {
-		configs[i] = bc.CopyWithPercentage(p)
-	}
-	return configs, nil
-}

lib/scheduler/configmap.go

@@ -50,9 +50,19 @@ func GetParsedConfig(name, configType string, rawJSON []byte) (result Config, er
 
 // UnmarshalJSON implements the json.Unmarshaler interface in a two-step manner,
 // creating the correct type of configs based on the `type` property.
-func (scs *ConfigMap) UnmarshalJSON(b []byte) error {
+func (scs *ConfigMap) UnmarshalJSON(data []byte) error {
+	if len(data) == 0 {
+		return nil
+	}
+
+	if len(data) == 4 && string(data) == "null" {
+		return nil
+	}
+
+	//TODO: use a more sophisticated combination of dec.Token() and dec.More(),
+	// which would allow us to support both arrays and maps for this config?
 	var protoConfigs map[string]protoConfig
-	if err := json.Unmarshal(b, &protoConfigs); err != nil {
+	if err := strictJSONUnmarshal(data, &protoConfigs); err != nil {
 		return err
 	}
 

lib/scheduler/constant_arrival_rate.go

@@ -1,7 +1,6 @@
 package scheduler
 
 import (
-	"encoding/json"
 	"fmt"
 	"time"
 
@@ -14,7 +13,7 @@ const constantArrivalRateType = "constant-arrival-rate"
 func init() {
 	RegisterConfigType(constantArrivalRateType, func(name string, rawJSON []byte) (Config, error) {
 		config := NewConstantArrivalRateConfig(name)
-		err := json.Unmarshal(rawJSON, &config)
+		err := strictJSONUnmarshal(rawJSON, &config)
 		return config, err
 	})
 }
@@ -23,7 +22,7 @@ func init() {
 type ConstantArrivalRateConfig struct {
 	BaseConfig
 	Rate     null.Int           `json:"rate"`
-	TimeUnit types.NullDuration `json:"timeUnit"` //TODO: rename to something else?
+	TimeUnit types.NullDuration `json:"timeUnit"`
 	Duration types.NullDuration `json:"duration"`
 
 	// Initialize `PreAllocatedVUs` number of VUs, and if more than that are needed,
@@ -38,7 +37,6 @@ func NewConstantArrivalRateConfig(name string) ConstantArrivalRateConfig {
 	return ConstantArrivalRateConfig{
 		BaseConfig: NewBaseConfig(name, constantArrivalRateType, false),
 		TimeUnit:   types.NewNullDuration(1*time.Second, false),
-		//TODO: set some default values for PreAllocatedVUs and MaxVUs?
 	}
 }
 
@@ -54,7 +52,7 @@ func (carc ConstantArrivalRateConfig) Validate() []error {
 		errors = append(errors, fmt.Errorf("the iteration rate should be positive"))
 	}
 
-	if time.Duration(carc.TimeUnit.Duration) < 0 {
+	if time.Duration(carc.TimeUnit.Duration) <= 0 {
 		errors = append(errors, fmt.Errorf("the timeUnit should be more than 0"))
 	}
 
@@ -80,3 +78,18 @@ func (carc ConstantArrivalRateConfig) Validate() []error {
 
 	return errors
 }
+
+// GetMaxVUs returns the absolute maximum number of possible concurrently running VUs
+func (carc ConstantArrivalRateConfig) GetMaxVUs() int64 {
+	return carc.MaxVUs.Int64
+}
+
+// GetMaxDuration returns the maximum duration time for this scheduler, including
+// the specified iterationTimeout, if the iterations are uninterruptible
+func (carc ConstantArrivalRateConfig) GetMaxDuration() time.Duration {
+	maxDuration := carc.Duration.Duration
+	if !carc.Interruptible.Bool {
+		maxDuration += carc.IterationTimeout.Duration
+	}
+	return time.Duration(maxDuration)
+}