New executors

cmd/cloud.go

@@ -23,6 +23,7 @@ package cmd
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -33,6 +34,7 @@ import (
 	"github.com/loadimpact/k6/lib"
 	"github.com/loadimpact/k6/stats/cloud"
 	"github.com/loadimpact/k6/ui"
+	"github.com/loadimpact/k6/ui/pb"
 	"github.com/pkg/errors"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
@@ -57,11 +59,9 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 	RunE: func(cmd *cobra.Command, args []string) error {
 		//TODO: disable in quiet mode?
 		_, _ = BannerColor.Fprintf(stdout, "\n%s\n\n", Banner)
-		initBar := ui.ProgressBar{
-			Width: 60,
-			Left:  func() string { return "    uploading script" },
-		}
-		fprintf(stdout, "%s \r", initBar.String())
+
+		progressBar := pb.New(pb.WithConstLeft(" Init"))
+		printBar(progressBar, "Parsing script")
 
 		// Runner
 		pwd, err := os.Getwd()
@@ -81,11 +81,13 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 			return err
 		}
 
+		printBar(progressBar, "Getting script options")
 		r, err := newRunner(src, runType, fs, runtimeOptions)
 		if err != nil {
 			return err
 		}
 
+		printBar(progressBar, "Consolidating options")
 		cliOpts, err := getOptions(cmd.Flags())
 		if err != nil {
 			return err
@@ -113,6 +115,7 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 			return errors.New("Not logged in, please use `k6 login cloud`.")
 		}
 
+		printBar(progressBar, "Building the archive")
 		arc := r.MakeArchive()
 		// TODO: Fix this
 		// We reuse cloud.Config for parsing options.ext.loadimpact, but this probably shouldn't be
@@ -161,22 +164,28 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 		}
 
 		// Start cloud test run
+		printBar(progressBar, "Validating script options")
 		client := cloud.NewClient(cloudConfig.Token.String, cloudConfig.Host.String, Version)
 		if err := client.ValidateOptions(arc.Options); err != nil {
 			return err
 		}
 
+		printBar(progressBar, "Uploading archive")
 		refID, err := client.StartCloudTestRun(name, cloudConfig.ProjectID.Int64, arc)
 		if err != nil {
 			return err
 		}
+		progressBar.Modify(pb.WithConstLeft("   Run"))
+		printBar(progressBar, "Initializing the cloud test")
 
 		testURL := cloud.URLForResults(refID, cloudConfig)
 		fprintf(stdout, "\n\n")
-		fprintf(stdout, "     execution: %s\n", ui.ValueColor.Sprint("cloud"))
+		fprintf(stdout, "   executor: %s\n", ui.ValueColor.Sprint("cloud"))
 		fprintf(stdout, "     script: %s\n", ui.ValueColor.Sprint(filename))
 		fprintf(stdout, "     output: %s\n", ui.ValueColor.Sprint(testURL))
+		//TODO: print schedulers information
 		fprintf(stdout, "\n")
+		printBar(progressBar, "Initializing the cloud test")
 
 		// The quiet option hides the progress bar and disallow aborting the test
 		if quiet {
@@ -190,12 +199,15 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 
 		var progressErr error
 		testProgress := &cloud.TestProgressResponse{}
-		progress := ui.ProgressBar{
-			Width: 60,
-			Left: func() string {
-				return "  " + testProgress.RunStatusText
-			},
+		percentageFmt := "[" + pb.GetFixedLengthFloatFormat(100, 2) + "%%] %s"
+		progressBar.Modify(
+			pb.WithProgress(func() (float64, string) {
+				if testProgress.RunStatus < lib.RunStatusRunning {
+					return 0, testProgress.RunStatusText
 		}
+				return testProgress.Progress, fmt.Sprintf(percentageFmt, testProgress.Progress*100, testProgress.RunStatusText)
+			}),
+		)
 
 		ticker := time.NewTicker(time.Millisecond * 2000)
 		shouldExitLoop := false
@@ -209,8 +221,7 @@ This will execute the test on the Load Impact cloud service. Use "k6 login cloud
 					if (testProgress.RunStatus > lib.RunStatusRunning) || (exitOnRunning && testProgress.RunStatus == lib.RunStatusRunning) {
 						shouldExitLoop = true
 					}
-					progress.Progress = testProgress.Progress
-					fprintf(stdout, "%s\x1b[0K\r", progress.String())
+					printBar(progressBar, "")
 				} else {
 					log.WithError(progressErr).Error("Test progress error")
 				}

cmd/collectors.go

@@ -60,7 +60,10 @@ func parseCollector(s string) (t, arg string) {
 	}
 }
 
-func newCollector(collectorName, arg string, src *lib.SourceData, conf Config) (lib.Collector, error) {
+func newCollector(
+	collectorName, arg string, src *lib.SourceData, conf Config, executionPlan []lib.ExecutionStep,
+) (lib.Collector, error) {
+
 	getCollector := func() (lib.Collector, error) {
 		switch collectorName {
 		case collectorJSON:
@@ -84,7 +87,7 @@ func newCollector(collectorName, arg string, src *lib.SourceData, conf Config) (
 			if arg != "" {
 				config.Name = null.StringFrom(arg)
 			}
-			return cloud.New(config, src, conf.Options, Version)
+			return cloud.New(config, src, conf.Options, executionPlan, Version)
 		case collectorKafka:
 			config := kafka.NewConfig().Apply(conf.Collectors.Kafka)
 			if err := envconfig.Process("k6", &config); err != nil {

cmd/common.go

@@ -21,11 +21,9 @@
 package cmd
 
 import (
-	"bytes"
 	"fmt"
 	"io"
 	"os"
-	"sync"
 
 	"github.com/loadimpact/k6/lib/types"
 	"github.com/spf13/afero"
@@ -51,23 +49,6 @@ type ExitCode struct {
 	Code int
 }
 
-// A writer that syncs writes with a mutex and, if the output is a TTY, clears before newlines.
-type consoleWriter struct {
-	Writer io.Writer
-	IsTTY  bool
-	Mutex  *sync.Mutex
-}
-
-func (w consoleWriter) Write(p []byte) (n int, err error) {
-	if w.IsTTY {
-		p = bytes.Replace(p, []byte{'\n'}, []byte{'\x1b', '[', '0', 'K', '\n'}, -1)
-	}
-	w.Mutex.Lock()
-	n, err = w.Writer.Write(p)
-	w.Mutex.Unlock()
-	return
-}
-
 //TODO: refactor the CLI config so these functions aren't needed - they
 // can mask errors by failing only at runtime, not at compile time
 func getNullBool(flags *pflag.FlagSet, key string) null.Bool {

cmd/config.go

@@ -37,7 +37,6 @@ import (
 	"github.com/loadimpact/k6/stats/influxdb"
 	"github.com/loadimpact/k6/stats/kafka"
 	"github.com/loadimpact/k6/stats/statsd/common"
-	log "github.com/sirupsen/logrus"
 	"github.com/spf13/afero"
 	"github.com/spf13/pflag"
 	null "gopkg.in/guregu/null.v3"
@@ -73,6 +72,14 @@ type Config struct {
 	} `json:"collectors"`
 }
 
+// Validate checks if all of the specified options make sense
+func (c Config) Validate() []error {
+	errors := c.Options.Validate()
+	//TODO: validate all of the other options... that we should have already been validating...
+	//TODO: maybe integrate an external validation lib: https://github.com/avelino/awesome-go#validation
+	return errors
+}
+
 func (c Config) Apply(cfg Config) Config {
 	c.Options = c.Options.Apply(cfg.Options)
 	if len(cfg.Out) > 0 {
@@ -179,100 +186,6 @@ func readEnvConfig() (conf Config, err error) {
 	return conf, nil
 }
 
-type executionConflictConfigError string
-
-func (e executionConflictConfigError) Error() string {
-	return string(e)
-}
-
-var _ error = executionConflictConfigError("")
-
-// 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
-	switch {
-	case conf.Duration.Valid:
-		if conf.Iterations.Valid {
-			//TODO: make this an executionConflictConfigError in the next version
-			log.Warnf("Specifying both duration and iterations is deprecated and won't be supported in the future k6 versions")
-		}
-
-		if len(conf.Stages) > 0 { // stages isn't nil (not set) and isn't explicitly set to empty
-			//TODO: make this an executionConflictConfigError 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, executionConflictConfigError("specifying both duration and execution is not supported")
-		}
-
-		if conf.Duration.Duration <= 0 {
-			//TODO: make this an executionConflictConfigError in the next version
-			log.Warnf("Specifying infinite duration in this way is deprecated and won't be supported in the future k6 versions")
-		} else {
-			ds := scheduler.NewConstantLoopingVUsConfig(lib.DefaultSchedulerName)
-			ds.VUs = conf.VUs
-			ds.Duration = conf.Duration
-			ds.Interruptible = null.NewBool(true, false) // Preserve backwards compatibility
-			result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
-		}
-
-	case len(conf.Stages) > 0: // stages isn't nil (not set) and isn't explicitly set to empty
-		if conf.Iterations.Valid {
-			//TODO: make this an executionConflictConfigError 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, executionConflictConfigError("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})
-			}
-		}
-		ds.Interruptible = null.NewBool(true, false) // Preserve backwards compatibility
-		result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
-
-	case conf.Iterations.Valid:
-		if conf.Execution != nil {
-			return conf, executionConflictConfigError("specifying both iterations and execution is not supported")
-		}
-		// TODO: maybe add a new flag that will be used as a shortcut to per-VU iterations?
-
-		ds := scheduler.NewSharedIterationsConfig(lib.DefaultSchedulerName)
-		ds.VUs = conf.VUs
-		ds.Iterations = conf.Iterations
-		result.Execution = scheduler.ConfigMap{lib.DefaultSchedulerName: ds}
-
-	default:
-		if conf.Execution != nil { // If someone set this, regardless if its empty
-			//TODO: remove this warning in the next version
-			log.Warnf("The execution settings are not functional in this k6 release, they will be ignored")
-		}
-
-		if len(conf.Execution) == 0 { // If unset or set to empty
-			// No execution parameters whatsoever were specified, so we'll create a per-VU iterations config
-			// with 1 VU and 1 iteration. We're choosing the per-VU config, since that one could also
-			// be executed both locally, and in the cloud.
-			result.Execution = scheduler.ConfigMap{
-				lib.DefaultSchedulerName: scheduler.NewPerVUIterationsConfig(lib.DefaultSchedulerName),
-			}
-		}
-	}
-
-	//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
@@ -302,11 +215,11 @@ func getConsolidatedConfig(fs afero.Fs, cliConf Config, runner lib.Runner) (conf
 	}
 	conf = conf.Apply(envConf).Apply(cliConf)
 
-	return buildExecutionConfig(conf)
+	conf.Options, err = scheduler.BuildExecutionConfig(conf.Options)
+
+	return conf, err
 }
 
-//TODO: remove ↓
-//nolint:unparam
 func validateConfig(conf Config) error {
 	errList := conf.Validate()
 	if len(errList) == 0 {
@@ -317,9 +230,6 @@ func validateConfig(conf Config) error {
 	for _, err := range errList {
 		errMsgParts = append(errMsgParts, fmt.Sprintf("\t- %s", err.Error()))
 	}
-	errMsg := errors.New(strings.Join(errMsgParts, "\n"))
 
-	//TODO: actually return the error here instead of warning, so k6 aborts on config validation errors
-	log.Warn(errMsg)
-	return nil
+	return errors.New(strings.Join(errMsgParts, "\n"))
 }