diff --git a/CHANGELOG.md b/CHANGELOG.md index 13c4dc2..93b828d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +- `experiment run --run-again` starts another run of an experiment that is already + running; without it, the run in progress is followed, as before. Each run it starts + sends its own `Idempotency-Key`, kept across the retries of that run, so a retried + request never starts yet another one. Needs a platform that knows `allowRunningAgain`. +- `schedule create` and `schedule update` take `--run-again` and `--no-run-again`, so a + schedule runs even when its experiment is still running, and `schedule list` shows it. + ## v6.3.0 - `user list` lists the users of the tenant, with their role, teams, what manages them diff --git a/README.md b/README.md index 38609f1..ccc14d8 100644 --- a/README.md +++ b/README.md @@ -361,6 +361,7 @@ canceled or errored it before it ran. A few options make it fit pipelines: | `--expect-reason "…"` | Also requires the run's reason to be exactly this | | `--expectation-retries 2` | Runs the experiment again when a run did not end as expected | | `--busy-retries 3` | Waits and tries again while another experiment runs, instead of failing | +| `--run-again` | Starts another run when the experiment is already running | | `--external-id shop-latency` | Runs the experiment with this external id, instead of `-k` | In GitHub Actions a summary of every run is added to the job summary. diff --git a/go.mod b/go.mod index 40b63a3..94b8e81 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ toolchain go1.27.3 tool github.com/oapi-codegen/oapi-codegen/v2/cmd/oapi-codegen require ( + github.com/google/uuid v1.6.0 github.com/itchyny/gojq v0.12.19 github.com/mattn/go-runewidth v0.0.30 github.com/oapi-codegen/runtime v1.7.0 @@ -25,7 +26,6 @@ require ( github.com/dprotaso/go-yit v0.0.0-20220510233725-9ba8df137936 // indirect github.com/getkin/kin-openapi v0.149.0 // indirect github.com/go-openapi/jsonpointer v1.0.2 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/itchyny/timefmt-go v0.1.8 // indirect github.com/oapi-codegen/oapi-codegen/v2 v2.8.0 // indirect diff --git a/internal/cli/experiment.go b/internal/cli/experiment.go index 7c32764..4cb2531 100644 --- a/internal/cli/experiment.go +++ b/internal/cli/experiment.go @@ -77,6 +77,7 @@ func newExperimentRun() *cobra.Command { "steadybit experiment run -f ./experiments -R --yes --parallel 3 --report steadybit.xml", "steadybit experiment run --template d7e65100-1d20-4980-be87-c351704910b8 --team ADM -p CLUSTER=prod", "steadybit experiment run --external-id shop-latency --yes --expect-state FAILED --busy-retries 3", + "steadybit experiment run -k ADM-1 --run-again", ), RunE: withClient(func(ctx context.Context, c *platform.Client, _ []string) error { o.Wait = !noWait @@ -90,6 +91,7 @@ func newExperimentRun() *cobra.Command { f.BoolVar(&noWait, "no-wait", false, "Do not wait for experiment run to finish. The run is still checked until it started, for up to 15 seconds, and a run the platform canceled or errored before it ran fails.") f.BoolVar(&o.Yes, "yes", false, "Skip the prompt asking for experiment run confirmation. Not necessary when no TTY is attached.") f.BoolVar(&o.AllowParallel, "allowParallel", false, "Skip the prompt warning about another experiment running and allow always parallel execution.") + f.BoolVar(&o.RunAgain, "run-again", false, "Start another run even when this experiment is already running. Without it, a running experiment's run in progress is followed instead.") f.IntVar(&o.Retries, "retries", 0, "Number of retries when the experiment fails validation (e.g., missing targets). 0 means no retry.") f.IntVar(&o.RetryInterval, "retryInterval", 10, "Interval in seconds between retries.") f.IntVar(&o.BusyRetries, "busy-retries", 0, "When another experiment is running and running in parallel is not allowed: try again this many times instead of asking, or failing in a pipeline.") diff --git a/internal/cli/schedule.go b/internal/cli/schedule.go index 23f5ad4..e2f81d4 100644 --- a/internal/cli/schedule.go +++ b/internal/cli/schedule.go @@ -21,28 +21,39 @@ func scheduleIDFlag(cmd *cobra.Command, id *string) { } // scheduleFields adds the flags `create` and `update` share. --allow-parallel and -// --no-allow-parallel stay unset unless one is given, so an update leaves it alone. +// --no-allow-parallel, like --run-again and --no-run-again, stay unset unless one is given, +// so an update leaves it alone. func scheduleFields(cmd *cobra.Command, f *schedule.Fields) { vars := newKeyValues() f.Variables = vars.values - var allow, noAllow bool + var allow, noAllow, again, noAgain bool flags := cmd.Flags() flags.StringVar(&f.Cron, "cron", "", `Run repeatedly on this Quartz cron expression (seconds first), e.g. "0 0 9 ? * MON-FRI".`) flags.StringVar(&f.StartAt, "start-at", "", "Run once at this ISO 8601 time, e.g. 2026-10-01T09:00:00Z.") flags.StringVar(&f.Timezone, "timezone", "", "The timezone of the cron expression, e.g. Europe/Berlin.") flags.BoolVar(&allow, "allow-parallel", false, "Run even when another experiment is running.") flags.BoolVar(&noAllow, "no-allow-parallel", false, "Skip the run when another experiment is running.") + flags.BoolVar(&again, "run-again", false, "Run even when this experiment is still running.") + flags.BoolVar(&noAgain, "no-run-again", false, "Skip the run when this experiment is still running.") flags.Var(vars, "variable", "A variable for the scheduled runs, overriding experiment and environment variables. Repeat for more.") cmd.MarkFlagsMutuallyExclusive("allow-parallel", "no-allow-parallel") + cmd.MarkFlagsMutuallyExclusive("run-again", "no-run-again") cmd.PreRun = func(*cobra.Command, []string) { - switch { - case allow: - f.AllowParallel = &allow - case noAllow: - no := false - f.AllowParallel = &no - } + f.AllowParallel = eitherFlag(allow, noAllow) + f.AllowRunningAgain = eitherFlag(again, noAgain) + } +} + +// eitherFlag is true for the flag, false for its --no- counterpart, and unset for neither. +func eitherFlag(yes, no bool) *bool { + switch { + case yes: + return &yes + case no: + value := false + return &value } + return nil } func newSchedule() *cobra.Command { diff --git a/internal/experiment/experiment.go b/internal/experiment/experiment.go index 8c4d6e0..c3a47e8 100644 --- a/internal/experiment/experiment.go +++ b/internal/experiment/experiment.go @@ -29,6 +29,7 @@ import ( "sync/atomic" "time" + "github.com/google/uuid" "github.com/steadybit/cli/v6/api" "github.com/steadybit/cli/v6/internal/interrupt" "github.com/steadybit/cli/v6/internal/jsyaml" @@ -270,6 +271,10 @@ type RunOptions struct { Recursive bool Yes, Wait bool AllowParallel bool + // RunAgain starts another run of an experiment that is already running, instead of + // following the run in progress. The platform then needs an Idempotency-Key, one per + // run started, so that a retried request never starts yet another run. + RunAgain bool Retries int RetryInterval int // Parallel is how many runs go at once; 0 or 1 runs them one after another. @@ -311,7 +316,7 @@ func Run(ctx context.Context, c *platform.Client, o RunOptions) error { case o.Template != "" && o.Key != "": return errors.New("--key cannot be combined with --template. Use `experiment apply --template -k` to update it.") case o.Template != "": - runs = append(runs, runner{"this one", func(parallel, persist bool) (started, error) { return runTemplate(ctx, c, o, parallel, persist) }, false}) + runs = append(runs, runner{"this one", func(a attempt) (started, error) { return runTemplate(ctx, c, o, a) }, false}) case len(o.Files) > 0: files, err := ResolveFiles(o.Files, o.Recursive) if err != nil { @@ -321,13 +326,13 @@ func Run(ctx context.Context, c *platform.Client, o RunOptions) error { return errors.New("If --key is specified, at most one --file can be specified.") } for _, file := range files { - runs = append(runs, runner{fileExperimentName(file), func(parallel, persist bool) (started, error) { return runFile(ctx, c, o, file, parallel, persist) }, false}) + runs = append(runs, runner{fileExperimentName(file), func(a attempt) (started, error) { return runFile(ctx, c, o, file, a) }, false}) } case o.Key != "": if o.ExternalID != "" { return errors.New("--external-id finds the experiment to run; leave out --key.") } - runs = append(runs, runner{o.Key, func(parallel, persist bool) (started, error) { return runKey(ctx, c, o.Key, parallel, persist) }, true}) + runs = append(runs, runner{o.Key, func(a attempt) (started, error) { return runKey(ctx, c, o.Key, a) }, true}) case o.ExternalID != "": // Without --template, the external id names an experiment that exists, as the // run-experiment action's externalId does. @@ -335,7 +340,7 @@ func Run(ctx context.Context, c *platform.Client, o RunOptions) error { if err != nil { return err } - runs = append(runs, runner{key, func(parallel, persist bool) (started, error) { return runKey(ctx, c, key, parallel, persist) }, true}) + runs = append(runs, runner{key, func(a attempt) (started, error) { return runKey(ctx, c, key, a) }, true}) default: return errors.New("Either --key, --file or --template must be specified.") } @@ -523,22 +528,45 @@ func runConcurrently(ctx context.Context, c *platform.Client, o RunOptions, runs // to start it, and where its key goes in what is printed. type runner struct { what string - // run starts it; persist keeps a run the platform refused, so that its errors can be - // seen in the platform. - run func(parallel, persist bool) (started, error) + // run starts it, as the attempt says. + run func(attempt) (started, error) keyLast bool } +// attempt is how one request to start a run goes out: parallel runs alongside another +// experiment, persist keeps a run the platform refused, so that its errors can be seen in +// the platform, and runAgain alongside the experiment's own runs, under idempotencyKey. +type attempt struct { + parallel, persist bool + runAgain bool + idempotencyKey string +} + +// params is what every run endpoint takes from an attempt. The key goes along only when +// running again, which requires one. +func (a attempt) params() (*bool, *bool, *bool, *string) { + parallel, persist, runAgain := a.parallel, a.persist, a.runAgain + if !a.runAgain { + return ¶llel, &persist, nil, nil + } + key := a.idempotencyKey + return ¶llel, &persist, &runAgain, &key +} + // withRetries retries validation errors, which clear up once targets appear, and offers // a parallel run when another experiment is already running, or, with --busy-retries, // waits for it instead. Only the last attempt at a run with validation errors is kept on // the platform, so that it shows what was wrong without the attempts before it. // The prefix starts its messages, telling runs apart when several start at once. -func withRetries(o RunOptions, what, prefix string, run func(parallel, persist bool) (started, error)) (started, error) { +func withRetries(o RunOptions, what, prefix string, run func(attempt) (started, error)) (started, error) { parallel := o.AllowParallel + // One key for the run being started, kept across the retries of it: none of them + // started a run, so the key is free, and a retry that did reach the platform before + // gets that run back rather than a second one. + idempotencyKey := uuid.NewString() busy := 0 for attempt := 0; ; attempt++ { - result, err := run(parallel, attempt >= o.Retries) + result, err := run(attemptOf(o, parallel, attempt >= o.Retries, idempotencyKey)) if err == nil { return result, nil } @@ -581,6 +609,10 @@ func withRetries(o RunOptions, what, prefix string, run func(parallel, persist b } } +func attemptOf(o RunOptions, parallel, persist bool, idempotencyKey string) attempt { + return attempt{parallel: parallel, persist: persist, runAgain: o.RunAgain, idempotencyKey: idempotencyKey} +} + func decodeStarted(body []byte, resp *http.Response, fallbackKey string) (started, error) { var r api.ExecuteExperimentResponseAO if len(body) > 0 { @@ -598,16 +630,17 @@ func decodeStarted(body []byte, resp *http.Response, fallbackKey string) (starte return s, nil } -func runKey(ctx context.Context, c *platform.Client, key string, parallel, persist bool) (started, error) { - body, resp, err := platform.Read(c.ExecuteExperimentWithBody(ctx, key, - &api.ExecuteExperimentParams{AllowParallel: ¶llel, ForcePersist: &persist}, "application/json", nil)) +func runKey(ctx context.Context, c *platform.Client, key string, a attempt) (started, error) { + parallel, persist, runAgain, idempotencyKey := a.params() + body, resp, err := platform.Read(c.ExecuteExperimentWithBody(ctx, key, &api.ExecuteExperimentParams{AllowParallel: parallel, + AllowRunningAgain: runAgain, IdempotencyKey: idempotencyKey, ForcePersist: persist}, "application/json", nil)) if err != nil { return started{}, err } return decodeStarted(body, resp, key) } -func runFile(ctx context.Context, c *platform.Client, o RunOptions, file string, parallel, persist bool) (started, error) { +func runFile(ctx context.Context, c *platform.Client, o RunOptions, file string, a attempt) (started, error) { document, datatype, err := load(file) if err != nil { return started{}, err @@ -620,14 +653,15 @@ func runFile(ctx context.Context, c *platform.Client, o RunOptions, file string, if err := update(ctx, c, key, document); err != nil { return started{}, err } - return runKey(ctx, c, key, parallel, persist) + return runKey(ctx, c, key, a) } reqBody, err := jsonBody(document) if err != nil { return started{}, err } - body, resp, err := platform.Read(c.SaveAndRunWithBody(ctx, - &api.SaveAndRunParams{AllowParallel: ¶llel, ForcePersist: &persist}, "application/json", reqBody)) + parallel, persist, runAgain, idempotencyKey := a.params() + body, resp, err := platform.Read(c.SaveAndRunWithBody(ctx, &api.SaveAndRunParams{AllowParallel: parallel, + AllowRunningAgain: runAgain, IdempotencyKey: idempotencyKey, ForcePersist: persist}, "application/json", reqBody)) if err != nil { return started{}, err } @@ -638,7 +672,7 @@ func runFile(ctx context.Context, c *platform.Client, o RunOptions, file string, return result, writeBack(file, document, datatype, result.Key) } -func runTemplate(ctx context.Context, c *platform.Client, o RunOptions, parallel, persist bool) (started, error) { +func runTemplate(ctx context.Context, c *platform.Client, o RunOptions, a attempt) (started, error) { id, err := templateID(o.Template) if err != nil { return started{}, err @@ -652,8 +686,9 @@ func runTemplate(ctx context.Context, c *platform.Client, o RunOptions, parallel Placeholders: create.Placeholders, ExperimentVariables: create.ExperimentVariables, ExecutionVariables: variables(o.ExecutionVariable), } - body, resp, err := platform.Read(c.SaveAndRunFromTemplate(ctx, id, - &api.SaveAndRunFromTemplateParams{ResetProperties: &o.ResetProperties, AllowParallel: ¶llel, ForcePersist: &persist}, request)) + parallel, persist, runAgain, idempotencyKey := a.params() + body, resp, err := platform.Read(c.SaveAndRunFromTemplate(ctx, id, &api.SaveAndRunFromTemplateParams{ResetProperties: &o.ResetProperties, + AllowParallel: parallel, AllowRunningAgain: runAgain, IdempotencyKey: idempotencyKey, ForcePersist: persist}, request)) if platform.IsStatus(err, http.StatusNotFound) { return started{}, fmt.Errorf("Experiment template %s not found.", o.Template) } diff --git a/internal/experiment/experiment_test.go b/internal/experiment/experiment_test.go index bc2b2ae..7fbb7d3 100644 --- a/internal/experiment/experiment_test.go +++ b/internal/experiment/experiment_test.go @@ -229,6 +229,71 @@ func TestRunGivesUpAfterTheLastRetry(t *testing.T) { assert.Len(t, p.Requests("POST /api/experiments/TST-1/execute"), 2) } +func TestRunWithoutRunAgainSendsNeitherTheFlagNorAKey(t *testing.T) { + p := platformtest.New(t) + stillRunning(p) + p.Reply("POST /api/experiments/TST-1/execute", started(p, "TST-1", 1)) + + _, err := platformtest.Stdout(t, func() error { + return experiment.Run(ctx, p.Client, experiment.RunOptions{Key: "TST-1", Yes: true}) + }) + + require.NoError(t, err) + request := p.Requests("POST /api/experiments/TST-1/execute")[0] + // A running experiment's run in progress comes back, as it always has. + assert.NotContains(t, request.Query, "allowRunningAgain") + assert.Empty(t, request.Header.Get("Idempotency-Key")) +} + +func TestRunAgainKeepsItsKeyAcrossTheRetriesOfOneRun(t *testing.T) { + p := platformtest.New(t) + stillRunning(p) + var calls atomic.Int32 + p.Handle("POST /api/experiments/TST-1/execute", func(platformtest.Request) platformtest.Reply { + if calls.Add(1) <= 2 { + return platformtest.Reply{Status: http.StatusUnprocessableEntity, JSON: map[string]any{"title": "no targets"}} + } + return started(p, "TST-1", 1) + }) + + _, err := platformtest.Stdout(t, func() error { + return experiment.Run(ctx, p.Client, experiment.RunOptions{Key: "TST-1", Yes: true, Retries: 2, RunAgain: true}) + }) + + require.NoError(t, err) + requests := p.Requests("POST /api/experiments/TST-1/execute") + require.Len(t, requests, 3) + // The platform requires a key to run again; one per run, so a retry never starts a second one. + key := requests[0].Header.Get("Idempotency-Key") + assert.NotEmpty(t, key) + for _, r := range requests { + assert.Equal(t, []string{"true"}, r.Query["allowRunningAgain"]) + assert.Equal(t, key, r.Header.Get("Idempotency-Key")) + } +} + +func TestRunAgainUsesANewKeyForEachRunItStarts(t *testing.T) { + p := platformtest.New(t) + p.Reply("POST /api/experiments/TST-1/execute", started(p, "TST-1", 1)) + var polls atomic.Int32 + p.Handle("GET /api/experiments/executions/1", func(platformtest.Request) platformtest.Reply { + if polls.Add(1) <= 1 { + return runState("FAILED", "flaky", false) + } + return runState("COMPLETED", "", false) + }) + + _, err := platformtest.Stdout(t, func() error { + return experiment.Run(ctx, p.Client, experiment.RunOptions{Key: "TST-1", Yes: true, Wait: true, ExpectationRetries: 1, RunAgain: true}) + }) + + require.NoError(t, err) + requests := p.Requests("POST /api/experiments/TST-1/execute") + require.Len(t, requests, 2) + // Running the experiment again is a new run, not a retry of the one before. + assert.NotEqual(t, requests[0].Header.Get("Idempotency-Key"), requests[1].Header.Get("Idempotency-Key")) +} + func TestRunsInParallelWithYesWhenAnotherIsRunning(t *testing.T) { p := platformtest.New(t) stillRunning(p) diff --git a/internal/gitops/kinds.go b/internal/gitops/kinds.go index c4c18ca..f35ea27 100644 --- a/internal/gitops/kinds.go +++ b/internal/gitops/kinds.go @@ -103,7 +103,7 @@ var Experiment = Kind{ var Schedule = Kind{ Name: "experiment schedule", ReadOnly: []string{"editedBy", "lastUpdated", "nextExecution"}, - Defaults: map[string]any{"allowParallel": true}, + Defaults: map[string]any{"allowParallel": true, "allowRunningAgain": false}, Identity: "id", Remote: func(ctx context.Context, c *platform.Client, local *jsyaml.Map) (string, *jsyaml.Map, error) { id := str(local, "id") diff --git a/internal/schedule/schedule.go b/internal/schedule/schedule.go index 741bab0..7a741e6 100644 --- a/internal/schedule/schedule.go +++ b/internal/schedule/schedule.go @@ -55,6 +55,7 @@ func List(ctx context.Context, c *platform.Client, o ListOptions) error { Timezone *string `json:"timezone"` Enabled *bool `json:"enabled"` AllowParallel *bool `json:"allowParallel"` + RunAgain *bool `json:"allowRunningAgain"` } resp, err := c.GetAllSchedulesV2(ctx, &api.GetAllSchedulesV2Params{Team: optional(o.Teams), Experiment: optional(o.Experiments)}) var raw []json.RawMessage @@ -77,10 +78,12 @@ func List(ctx context.Context, c *platform.Client, o ListOptions) error { table.Column{Name: "when", Title: "When", Alignment: table.Left}, table.Column{Name: "enabled", Title: "Enabled", Alignment: table.Left}, table.Column{Name: "allowParallel", Title: "Parallel", Alignment: table.Left}, + table.Column{Name: "allowRunningAgain", Title: "Run again", Alignment: table.Left}, ) - boolOr := func(b *bool) string { + // Each as the platform defaults it when a schedule does not say. + boolOr := func(b *bool, fallback bool) string { if b == nil { - return "true" + return fmt.Sprint(fallback) } return fmt.Sprint(*b) } @@ -96,7 +99,8 @@ func List(ctx context.Context, c *platform.Client, o ListOptions) error { when = *s.StartAt } t.AddRow(table.Default, table.Cell("id", s.ID), table.Cell("experiment", s.ExperimentKey), table.Cell("when", when), - table.Cell("enabled", boolOr(s.Enabled)), table.Cell("allowParallel", boolOr(s.AllowParallel))) + table.Cell("enabled", boolOr(s.Enabled, true)), table.Cell("allowParallel", boolOr(s.AllowParallel, true)), + table.Cell("allowRunningAgain", boolOr(s.RunAgain, false))) } t.Print() return nil @@ -158,6 +162,7 @@ func save(ctx context.Context, c *platform.Client, schedule *jsyaml.Map) (upsert type Fields struct { Cron, StartAt, Timezone string AllowParallel *bool + AllowRunningAgain *bool Variables *jsyaml.Map } @@ -181,6 +186,9 @@ func (f Fields) into(m *jsyaml.Map) { if f.AllowParallel != nil { m.Set("allowParallel", *f.AllowParallel) } + if f.AllowRunningAgain != nil { + m.Set("allowRunningAgain", *f.AllowRunningAgain) + } if f.Variables != nil && f.Variables.Len() > 0 { m.Set("variables", f.Variables) } diff --git a/internal/schedule/schedule_test.go b/internal/schedule/schedule_test.go index b90d160..fa9c82c 100644 --- a/internal/schedule/schedule_test.go +++ b/internal/schedule/schedule_test.go @@ -102,6 +102,19 @@ func TestCreateAndItsRules(t *testing.T) { assert.EqualError(t, schedule.Create(ctx, p.Client, schedule.CreateOptions{Experiment: "ADM-1", Fields: schedule.Fields{Cron: "x", StartAt: "y"}}), "--cron and --start-at cannot be combined.") } +func TestCreateRunningAgain(t *testing.T) { + p := platformtest.New(t) + p.Reply("POST /api/experiments/schedules", platformtest.Reply{Status: http.StatusCreated, JSON: saved}) + again := true + + _, err := platformtest.Stdout(t, func() error { + return schedule.Create(ctx, p.Client, schedule.CreateOptions{Experiment: "ADM-1", Fields: schedule.Fields{Cron: "0 0 9 ? * *", AllowRunningAgain: &again}}) + }) + + require.NoError(t, err) + assert.Equal(t, true, p.Requests("POST /api/experiments/schedules")[0].JSON(t).(map[string]any)["allowRunningAgain"]) +} + func TestUpdateSendsOnlyWhatWasGiven(t *testing.T) { p := platformtest.New(t) p.Reply("PATCH /api/experiments/schedules/"+id, platformtest.Reply{JSON: saved})