Use a new interface for Outputs and adapt old Collectors to it (#1869)

This also should allow k6 to support xk6 extensions for outputs. So far the JSON output is the only one that has been fully transitioned to the new output.Output interface, the rest will be done in future PRs.

api/server_test.go

@@ -82,7 +82,7 @@ func TestWithEngine(t *testing.T) {
 	logger.SetOutput(testutils.NewTestOutput(t))
 	execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{}, logger)
 	require.NoError(t, err)
-	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, logger)
+	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, nil, logger)
 	require.NoError(t, err)
 
 	rw := httptest.NewRecorder()

api/v1/group_routes_test.go

@@ -51,7 +51,7 @@ func TestGetGroups(t *testing.T) {
 
 	execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{Group: g0}, logger)
 	require.NoError(t, err)
-	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, logger)
+	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, nil, logger)
 	require.NoError(t, err)
 
 	t.Run("list", func(t *testing.T) {

api/v1/metric_routes_test.go

@@ -45,7 +45,7 @@ func TestGetMetrics(t *testing.T) {
 	logger.SetOutput(testutils.NewTestOutput(t))
 	execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{}, logger)
 	require.NoError(t, err)
-	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, logger)
+	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, nil, logger)
 	require.NoError(t, err)
 
 	engine.Metrics = map[string]*stats.Metric{
@@ -88,7 +88,7 @@ func TestGetMetric(t *testing.T) {
 	logger.SetOutput(testutils.NewTestOutput(t))
 	execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{}, logger)
 	require.NoError(t, err)
-	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, logger)
+	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, nil, logger)
 	require.NoError(t, err)
 
 	engine.Metrics = map[string]*stats.Metric{

api/v1/setup_teardown_routes_test.go

@@ -155,7 +155,7 @@ func TestSetupData(t *testing.T) {
 			})
 			execScheduler, err := local.NewExecutionScheduler(runner, logger)
 			require.NoError(t, err)
-			engine, err := core.NewEngine(execScheduler, runner.GetOptions(), lib.RuntimeOptions{}, logger)
+			engine, err := core.NewEngine(execScheduler, runner.GetOptions(), lib.RuntimeOptions{}, nil, logger)
 			require.NoError(t, err)
 
 			globalCtx, globalCancel := context.WithCancel(context.Background())

api/v1/status_routes_test.go

@@ -47,7 +47,7 @@ func TestGetStatus(t *testing.T) {
 	logger.SetOutput(testutils.NewTestOutput(t))
 	execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{}, logger)
 	require.NoError(t, err)
-	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, logger)
+	engine, err := core.NewEngine(execScheduler, lib.Options{}, lib.RuntimeOptions{}, nil, logger)
 	require.NoError(t, err)
 
 	rw := httptest.NewRecorder()
@@ -101,7 +101,7 @@ func TestPatchStatus(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			execScheduler, err := local.NewExecutionScheduler(&minirunner.MiniRunner{Options: options}, logger)
 			require.NoError(t, err)
-			engine, err := core.NewEngine(execScheduler, options, lib.RuntimeOptions{}, logger)
+			engine, err := core.NewEngine(execScheduler, options, lib.RuntimeOptions{}, nil, logger)
 			require.NoError(t, err)
 			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 			defer cancel()

cloudapi/config.go

@@ -26,6 +26,7 @@ import (
 
 	"gopkg.in/guregu/null.v3"
 
+	"github.com/kelseyhightower/envconfig"
 	"github.com/loadimpact/k6/lib/types"
 )
 
@@ -241,7 +242,8 @@ func (c Config) Apply(cfg Config) Config {
 	return c
 }
 
-// MergeFromExternal merges three fields from json in a loadimpact key of the provided external map
+// MergeFromExternal merges three fields from the JSON in a loadimpact key of
+// the provided external map. Used for options.ext.loadimpact settings.
 func MergeFromExternal(external map[string]json.RawMessage, conf *Config) error {
 	if val, ok := external["loadimpact"]; ok {
 		// TODO: Important! Separate configs and fix the whole 2 configs mess!
@@ -262,3 +264,29 @@ func MergeFromExternal(external map[string]json.RawMessage, conf *Config) error
 	}
 	return nil
 }
+
+// GetConsolidatedConfig combines the default config values with the JSON config
+// values and environment variables and returns the final result.
+func GetConsolidatedConfig(jsonRawConf json.RawMessage, env map[string]string, configArg string) (Config, error) {
+	result := NewConfig()
+	if jsonRawConf != nil {
+		jsonConf := Config{}
+		if err := json.Unmarshal(jsonRawConf, &jsonConf); err != nil {
+			return result, err
+		}
+		result = result.Apply(jsonConf)
+	}
+
+	envConfig := Config{}
+	if err := envconfig.Process("", &envConfig); err != nil {
+		// TODO: get rid of envconfig and actually use the env parameter...
+		return result, err
+	}
+	result = result.Apply(envConfig)
+
+	if configArg != "" {
+		result.Name = null.StringFrom(configArg)
+	}
+
+	return result, nil
+}

cmd/cloud.go

@@ -33,7 +33,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/kelseyhightower/envconfig"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/afero"
@@ -105,7 +104,8 @@ This will execute the test on the k6 cloud service. Use "k6 login cloud" to auth
 				return err
 			}
 
-			runtimeOptions, err := getRuntimeOptions(cmd.Flags(), buildEnvMap(os.Environ()))
+			osEnvironment := buildEnvMap(os.Environ())
+			runtimeOptions, err := getRuntimeOptions(cmd.Flags(), osEnvironment)
 			if err != nil {
 				return err
 			}
@@ -141,8 +141,8 @@ This will execute the test on the k6 cloud service. Use "k6 login cloud" to auth
 			}
 
 			// Cloud config
-			cloudConfig := cloudapi.NewConfig().Apply(derivedConf.Collectors.Cloud)
-			if err = envconfig.Process("", &cloudConfig); err != nil {
+			cloudConfig, err := cloudapi.GetConsolidatedConfig(derivedConf.Collectors["cloud"], osEnvironment, "")
+			if err != nil {
 				return err
 			}
 			if !cloudConfig.Token.Valid {
@@ -153,8 +153,8 @@ This will execute the test on the k6 cloud service. Use "k6 login cloud" to auth
 			arc := r.MakeArchive()
 			// TODO: Fix this
 			// We reuse cloud.Config for parsing options.ext.loadimpact, but this probably shouldn't be
-			// done as the idea of options.ext is that they are extensible without touching k6. But in
-			// order for this to happen we shouldn't actually marshall cloud.Config on top of it because
+			// done, as the idea of options.ext is that they are extensible without touching k6. But in
+			// order for this to happen, we shouldn't actually marshall cloud.Config on top of it, because
 			// it will be missing some fields that aren't actually mentioned in the struct.
 			// So in order for use to copy the fields that we need for loadimpact's api we unmarshal in
 			// map[string]interface{} and copy what we need if it isn't set already