k6runner: handle errors reported by http runners

internal/k6runner/k6runner.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -74,6 +75,11 @@ func NewScript(script []byte, k6runner Runner) (*Script, error) {
 	return &r, nil
 }
 
+var (
+	ErrBuggyRunner = errors.New("runner returned buggy response")
+	ErrFromRunner  = errors.New("runner reported an error")
+)
+
 func (r Script) Run(ctx context.Context, registry *prometheus.Registry, logger logger.Logger, internalLogger zerolog.Logger) (bool, error) {
 	k6runner := r.runner.WithLogger(&internalLogger)
 
@@ -111,7 +117,29 @@ func (r Script) Run(ctx context.Context, registry *prometheus.Registry, logger l
 		return false, err
 	}
 
-	return !resultCollector.failure, nil
+	// If only one of Error and ErrorCode are non-empty, the proxy is misbehaving.
+	switch {
+	case result.Error == "" && result.ErrorCode != "":
+		fallthrough
+	case result.Error != "" && result.ErrorCode == "":
+		return false, fmt.Errorf(
+			"%w: only one of error (%q) and errorCode (%q) is non-empty",
+			ErrBuggyRunner, result.Error, result.ErrorCode,
+		)
+	}
+
+	// https://github.com/grafana/sm-k6-runner/blob/b811839d444a7e69fd056b0a4e6ccf7e914197f3/internal/mq/runner.go#L51
+	switch result.ErrorCode {
+	case "":
+		// No error, all good.
+		return true, nil
+	case "timeout", "killed", "user":
+		// These are user errors. The probe failed, but we don't return an error.
+		return false, nil
+	default:
+		// We got an "unknown" error, or some other code we do not recognize. Return it so we log it.
+		return false, fmt.Errorf("%w: %s: %s", ErrFromRunner, result.ErrorCode, result.Error)
+	}
 }
 
 type customCollector struct {
@@ -293,8 +321,10 @@ type RunRequest struct {
 }
 
 type RunResponse struct {
-	Metrics []byte `json:"metrics"`
-	Logs    []byte `json:"logs"`
+	Error     string `json:"error,omitempty"`
+	ErrorCode string `json:"errorCode,omitempty"`
+	Metrics   []byte `json:"metrics"`
+	Logs      []byte `json:"logs"`
 }
 
 func (r HttpRunner) WithLogger(logger *zerolog.Logger) Runner {
@@ -304,6 +334,8 @@ func (r HttpRunner) WithLogger(logger *zerolog.Logger) Runner {
 	}
 }
 
+var ErrUnexpectedStatus = errors.New("unexpected status code")
+
 func (r HttpRunner) Run(ctx context.Context, script []byte) (*RunResponse, error) {
 	req, err := json.Marshal(&RunRequest{
 		Script: script,
@@ -323,24 +355,17 @@ func (r HttpRunner) Run(ctx context.Context, script []byte) (*RunResponse, error
 
 	defer resp.Body.Close()
 
-	dec := json.NewDecoder(resp.Body)
-
-	if resp.StatusCode != http.StatusOK {
-		var result requestError
-
-		err := dec.Decode(&result)
-		if err != nil {
-			r.logger.Error().Err(err).Msg("decoding request response")
-			return nil, fmt.Errorf("running script: %w", err)
-		}
-
-		r.logger.Error().Err(result).Msg("request response")
-		return nil, fmt.Errorf("running script: %w", result)
+	switch resp.StatusCode {
+	case http.StatusOK, http.StatusRequestTimeout, http.StatusUnprocessableEntity, http.StatusInternalServerError:
+	// These are status code that come with a machine-readable response. The response may contain an error, which is
+	// handled later.
+	// See: https://github.com/grafana/sm-k6-runner/blob/main/internal/mq/proxy.go#L215
+	default:
+		return nil, fmt.Errorf("%w %d", ErrUnexpectedStatus, resp.StatusCode)
 	}
 
 	var result RunResponse
-
-	err = dec.Decode(&result)
+	err = json.NewDecoder(resp.Body).Decode(&result)
 	if err != nil {
 		r.logger.Error().Err(err).Msg("decoding script result")
 		return nil, fmt.Errorf("decoding script result: %w", err)

internal/k6runner/k6runner_test.go

@@ -164,6 +164,127 @@ func TestHttpRunnerRunError(t *testing.T) {
 	require.Error(t, err)
 }
 
+// TestScriptHTTPRun tests that Script reports what it should depending on the status code and responses of the HTTP
+// runner.
+func TestScriptHTTPRun(t *testing.T) {
+	t.Parallel()
+
+	var (
+		testMetrics = testhelper.MustReadFile(t, "testdata/test.out")
+		testLogs    = testhelper.MustReadFile(t, "testdata/test.log")
+	)
+
+	for _, tc := range []struct {
+		name          string
+		response      *RunResponse
+		statusCode    int
+		expectSuccess bool
+		expectError   error
+	}{
+		{
+			name: "all good",
+			response: &RunResponse{
+				Metrics: testMetrics,
+				Logs:    testLogs,
+			},
+			statusCode:    http.StatusOK,
+			expectSuccess: true,
+			expectError:   nil,
+		},
+		{
+			// HTTP runner should report failure and an error when the upstream status is not recognized.
+			name:          "unexpected status",
+			response:      &RunResponse{},
+			statusCode:    999,
+			expectSuccess: false,
+			expectError:   ErrUnexpectedStatus,
+		},
+		{
+			// HTTP runner should report failure and an error when the error is unknown.
+			name: "non-user error",
+			response: &RunResponse{
+				Metrics:   testMetrics,
+				Logs:      testLogs,
+				Error:     "something went wrong",
+				ErrorCode: "something-wrong",
+			},
+			statusCode:    http.StatusOK,
+			expectSuccess: false,
+			expectError:   ErrFromRunner,
+		},
+		{
+			// HTTP runner should report failure but no error when the error is unknown.
+			name: "user error",
+			response: &RunResponse{
+				Metrics:   testMetrics,
+				Logs:      testLogs,
+				Error:     "syntax error somewhere or something",
+				ErrorCode: "user",
+			},
+			statusCode:    http.StatusOK,
+			expectSuccess: false,
+			expectError:   nil,
+		},
+		{
+			name: "inconsistent runner response A",
+			response: &RunResponse{
+				Metrics:   testMetrics,
+				Logs:      testLogs,
+				Error:     "set",
+				ErrorCode: "",
+			},
+			statusCode:    http.StatusInternalServerError,
+			expectSuccess: false,
+			expectError:   ErrBuggyRunner,
+		},
+		{
+			name: "inconsistent runner response B",
+			response: &RunResponse{
+				Metrics:   testMetrics,
+				Logs:      testLogs,
+				Error:     "",
+				ErrorCode: "set",
+			},
+			statusCode:    http.StatusInternalServerError,
+			expectSuccess: false,
+			expectError:   ErrBuggyRunner,
+		},
+	} {
+		tc := tc
+
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			mux := http.NewServeMux()
+			mux.HandleFunc("/run", func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(tc.statusCode)
+				_ = json.NewEncoder(w).Encode(tc.response)
+			})
+			srv := httptest.NewServer(mux)
+			t.Cleanup(srv.Close)
+
+			runner := New(RunnerOpts{Uri: srv.URL + "/run"})
+			script, err := NewScript([]byte("tee-hee"), runner)
+			require.NoError(t, err)
+
+			ctx, cancel := testhelper.Context(context.Background(), t)
+			t.Cleanup(cancel)
+
+			var (
+				registry = prometheus.NewRegistry()
+				logger   testLogger
+				buf      bytes.Buffer
+				zlogger  = zerolog.New(&buf)
+			)
+
+			success, err := script.Run(ctx, registry, &logger, zlogger)
+			require.Equal(t, tc.expectSuccess, success)
+			require.ErrorIs(t, err, tc.expectError)
+		})
+	}
+}
+
 type testRunner struct {
 	metrics []byte
 	logs    []byte