Gracefully handle load driver iteration failures

cmd/omes/run_scenario.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"strings"
@@ -15,6 +16,11 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	iterationFailurePolicyContinue = "continue"
+	iterationFailurePolicyFailFast = "fail-fast"
+)
+
 func runScenarioCmd() *cobra.Command {
 	var r scenarioRunner
 	cmd := &cobra.Command{
@@ -51,6 +57,7 @@ type scenarioRunConfig struct {
 	maxConcurrent                 int
 	maxIterationsPerSecond        float64
 	maxIterationAttempts          int
+	iterationFailurePolicy        string
 	scenarioOptions               []string
 	timeout                       time.Duration
 	doNotRegisterSearchAttributes bool
@@ -75,6 +82,8 @@ func (r *scenarioRunConfig) addCLIFlags(fs *pflag.FlagSet) {
 	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.IntVar(&r.maxIterationAttempts, "max-iteration-attempts", 1, "Maximum attempts per iteration")
+	fs.StringVar(&r.iterationFailurePolicy, "iteration-failure-policy", iterationFailurePolicyContinue,
+		"How to handle terminal iteration failures: continue or fail-fast")
 	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")
@@ -109,6 +118,13 @@ func (r *scenarioRunner) validateInput() (*loadgen.Scenario, *loadgen.OptionSet,
 		return nil, nil, loadgen.NewUsageError("--iterations and --duration cannot be combined; " +
 			"use --iterations to run a fixed number of times, or --duration to keep starting " +
 			"iterations for a period")
+	} else if policy := r.resolvedIterationFailurePolicy(); policy != iterationFailurePolicyContinue && policy != iterationFailurePolicyFailFast {
+		return nil, nil, loadgen.NewUsageError(
+			"--iteration-failure-policy must be %q or %q, got %q",
+			iterationFailurePolicyContinue,
+			iterationFailurePolicyFailFast,
+			r.iterationFailurePolicy,
+		)
 	}
 
 	// Parse options
@@ -140,6 +156,43 @@ func (r *scenarioRunner) validateInput() (*loadgen.Scenario, *loadgen.OptionSet,
 	return scenario, resolvedOptions, nil
 }
 
+func (r scenarioRunConfig) resolvedIterationFailurePolicy() string {
+	if r.iterationFailurePolicy == "" {
+		return iterationFailurePolicyContinue
+	}
+	return r.iterationFailurePolicy
+}
+
+func (r scenarioRunConfig) loadgenConfiguration() loadgen.RunConfiguration {
+	return loadgen.RunConfiguration{
+		Iterations:                    r.iterations,
+		Duration:                      r.duration,
+		MaxConcurrent:                 r.maxConcurrent,
+		MaxIterationsPerSecond:        r.maxIterationsPerSecond,
+		MaxIterationAttempts:          r.maxIterationAttempts,
+		Timeout:                       r.timeout,
+		DoNotRegisterSearchAttributes: r.doNotRegisterSearchAttributes,
+		IgnoreAlreadyStarted:          r.ignoreAlreadyStarted,
+		ContinueOnIterationFailure:    r.resolvedIterationFailurePolicy() == iterationFailurePolicyContinue,
+	}
+}
+
+// iterationFailuresOnly recognizes an IterationFailuresError through ordinary
+// single-cause wrapping. It deliberately rejects multi-errors so a degraded
+// completion cannot hide another run-level failure joined to it.
+func iterationFailuresOnly(err error) (*loadgen.IterationFailuresError, bool) {
+	for err != nil {
+		if failures, ok := err.(*loadgen.IterationFailuresError); ok {
+			return failures, true
+		}
+		if _, ok := err.(interface{ Unwrap() []error }); ok {
+			return nil, false
+		}
+		err = errors.Unwrap(err)
+	}
+	return nil, false
+}
+
 func (r *scenarioRunner) run(ctx context.Context) error {
 	scenario, resolvedOptions, err := r.validateInput()
 	if err != nil {
@@ -192,19 +245,10 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 		MetricsHandler: metrics.NewHandler(),
 		Client:         client,
 		ClientOptions:  r.clientOptions,
-		Configuration: loadgen.RunConfiguration{
-			Iterations:                    r.iterations,
-			Duration:                      r.duration,
-			MaxConcurrent:                 r.maxConcurrent,
-			MaxIterationsPerSecond:        r.maxIterationsPerSecond,
-			MaxIterationAttempts:          r.maxIterationAttempts,
-			Timeout:                       r.timeout,
-			DoNotRegisterSearchAttributes: r.doNotRegisterSearchAttributes,
-			IgnoreAlreadyStarted:          r.ignoreAlreadyStarted,
-		},
-		Options:   resolvedOptions,
-		Namespace: r.clientOptions.Namespace,
-		RootPath:  repoDir,
+		Configuration:  r.loadgenConfiguration(),
+		Options:        resolvedOptions,
+		Namespace:      r.clientOptions.Namespace,
+		RootPath:       repoDir,
 		ExportOptions: loadgen.ExportOptions{
 			ExportHistoriesDir:    r.exportHistoriesDir,
 			ExportHistoriesFilter: r.exportHistoriesFilter,
@@ -213,7 +257,14 @@ func (r *scenarioRunner) run(ctx context.Context) error {
 	executor := scenario.ExecutorFn()
 	err = executor.Run(ctx, scenarioInfo)
 	if err != nil {
-		return fmt.Errorf("failed scenario: %w", err)
+		if r.resolvedIterationFailurePolicy() == iterationFailurePolicyContinue {
+			if _, ok := iterationFailuresOnly(err); ok {
+				err = nil
+			}
+		}
+		if err != nil {
+			return fmt.Errorf("failed scenario: %w", err)
+		}
 	}
 	err = loadgen.ExportWorkflowHistories(ctx, scenarioInfo)
 	if err != nil {

cmd/omes/run_scenario_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"errors"
+	"fmt"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -45,6 +46,11 @@ func TestValidateInputClassifiesBadInputAsUsageErrors(t *testing.T) {
 			},
 			wantMsg: "cannot be combined",
 		},
+		{
+			name:    "invalid iteration failure policy",
+			mutate:  func(r *scenarioRunner) { r.iterationFailurePolicy = "sometimes" },
+			wantMsg: "--iteration-failure-policy must be",
+		},
 		{
 			name:    "option without equals",
 			mutate:  func(r *scenarioRunner) { r.scenarioOptions = []string{"novalue"} },
@@ -85,6 +91,43 @@ func TestValidateInputClassifiesBadInputAsUsageErrors(t *testing.T) {
 	}
 }
 
+func TestIterationFailurePolicyConfiguration(t *testing.T) {
+	tests := []struct {
+		name   string
+		policy string
+		want   bool
+	}{
+		{name: "zero value defaults to continue", want: true},
+		{name: "continue", policy: iterationFailurePolicyContinue, want: true},
+		{name: "fail fast", policy: iterationFailurePolicyFailFast, want: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			config := scenarioRunConfig{iterationFailurePolicy: test.policy}.loadgenConfiguration()
+			if config.ContinueOnIterationFailure != test.want {
+				t.Fatalf("ContinueOnIterationFailure = %v, want %v", config.ContinueOnIterationFailure, test.want)
+			}
+		})
+	}
+}
+
+func TestIterationFailuresOnly(t *testing.T) {
+	degraded := &loadgen.IterationFailuresError{Attempted: 2, Succeeded: 1, Failed: 1}
+
+	found, ok := iterationFailuresOnly(fmt.Errorf("scenario wrapper: %w", degraded))
+	if !ok || found != degraded {
+		t.Fatalf("expected wrapped degraded completion to be recognized, got %v, %v", found, ok)
+	}
+
+	if _, ok := iterationFailuresOnly(errors.Join(degraded, errors.New("cleanup failed"))); ok {
+		t.Fatal("must not treat a joined run-level error as only iteration failures")
+	}
+	if _, ok := iterationFailuresOnly(errors.New("run failed")); ok {
+		t.Fatal("must not treat an ordinary run error as degraded completion")
+	}
+}
+
 func TestValidateInputAcceptsGoodInput(t *testing.T) {
 	r := newRunner("throughput_stress")
 	r.scenarioOptions = []string{"sleep-time=3s"}

docs/authoring-scenarios.md

@@ -143,7 +143,8 @@ scenario's `DefaultConfiguration` field over the older `HasDefaultConfiguration`
 
 Scenario configuration arrives through **two separate channels**, and knowing which is which matters:
 
-1. **Built-in run flags** — iterations, duration, concurrency, rate, attempts, timeout. These are
+1. **Built-in run flags** — iterations, duration, concurrency, rate, attempts, iteration-failure policy,
+   timeout. These are
    framework-level and apply to every scenario, so you neither declare nor read them; see
    [running.md](./running.md#configuring-the-load) for the list.
 2. **Your own options** — `--option key=value` pairs that you **declare** on the scenario.

docs/running.md

@@ -70,12 +70,20 @@ These apply to every scenario and override its defaults:
 | `--max-concurrent` | Max iterations running at once. |
 | `--max-iterations-per-second` | Rate limit on starting iterations (0 = unlimited). |
 | `--max-iteration-attempts` | Attempts per iteration (default 1). |
+| `--iteration-failure-policy` | `continue` (default) records terminal failures and keeps generating load; `fail-fast` stops on the first terminal failure. |
 | `--timeout` | Hard stop; cancels in-flight iterations and exits non-zero. |
 
 If you set neither `--iterations` nor `--duration`, the scenario's own default applies — and most
 scenarios declare none, in which case omes's default does. `list-scenarios` states which is the case for
 each scenario.
 
+Iteration retries and the terminal-failure policy are independent. `--max-iteration-attempts` controls
+how many times one logical iteration may execute; after those attempts are exhausted, the default
+`continue` policy records the iteration as failed and starts more load. A completed run logs attempted,
+succeeded, and failed totals plus success/failure rates and successful iterations per second. When the
+load-driver Prometheus endpoint is enabled with `--prom-listen-address`, the same terminal outcomes are
+exported as `omes_iterations_total`, labeled by scenario, outcome, and status code.
+
 ### 2. Per-scenario options (`--option key=value`)
 
 Scenario-specific knobs are passed as repeated `--option key=value` pairs. Each scenario declares the