From 9760660318ba616382783f3b9ac34e41378b9307 Mon Sep 17 00:00:00 2001 From: Thomas Mullaney Date: Sun, 28 Jun 2026 01:01:21 -0700 Subject: [PATCH 1/5] Make OMES test worker dirs unique --- loadgen/kitchen_sink_executor_test.go | 4 ++-- workers/test_env.go | 10 +++++++++- workers/test_env_test.go | 17 +++++++++++++++++ 3 files changed, 28 insertions(+), 3 deletions(-) create mode 100644 workers/test_env_test.go diff --git a/loadgen/kitchen_sink_executor_test.go b/loadgen/kitchen_sink_executor_test.go index 5865df3a..5f65ff1f 100644 --- a/loadgen/kitchen_sink_executor_test.go +++ b/loadgen/kitchen_sink_executor_test.go @@ -48,8 +48,8 @@ type testCase struct { // TestKitchenSink tests specific kitchensink features across SDKs. // Use the `SDK` environment variable to run only a specific SDK. func TestKitchenSink(t *testing.T) { - if os.Getenv("CI") != "" && onlySDK == "" { - t.Skip("Skipping kitchensink test in CI without specific SDK set") + if onlySDK == "" && os.Getenv("OMES_RUN_ALL_SDKS") == "" { + t.Skip("Skipping kitchensink test without specific SDK set; set SDK or OMES_RUN_ALL_SDKS=1") } env := SetupTestEnvironment(t) diff --git a/workers/test_env.go b/workers/test_env.go index 9663b2b3..f3c287eb 100644 --- a/workers/test_env.go +++ b/workers/test_env.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path/filepath" + "sync/atomic" "testing" "time" @@ -29,6 +30,8 @@ const ( workerShutdownTimeout = 5 * time.Second ) +var testBuildDirSequence atomic.Uint64 + // Functional options configuration type testEnvConfig struct { executorTimeout time.Duration @@ -62,6 +65,7 @@ type TestEnvironment struct { temporalClient client.Client baseDir string buildDir string + buildDirNameValue string repoDir string nexusEndpointName string workerPool *workerPool @@ -115,6 +119,7 @@ func SetupTestEnvironment(t *testing.T, opts ...TestEnvOption) *TestEnvironment repoDir: repoDir, createdAt: time.Now(), } + env.buildDirNameValue = fmt.Sprintf("omes-temp-%d-%d", env.createdAt.UnixMilli(), testBuildDirSequence.Add(1)) env.workerPool = NewWorkerPool(env) // Optionally create a Nexus endpoint @@ -213,5 +218,8 @@ func (env *TestEnvironment) RunExecutorTest( } func (env *TestEnvironment) buildDirName() string { - return fmt.Sprintf("omes-temp-%d", env.createdAt.UnixMilli()) + if env.buildDirNameValue == "" { + env.buildDirNameValue = fmt.Sprintf("omes-temp-%d-%d", env.createdAt.UnixMilli(), testBuildDirSequence.Add(1)) + } + return env.buildDirNameValue } diff --git a/workers/test_env_test.go b/workers/test_env_test.go new file mode 100644 index 00000000..e2216c84 --- /dev/null +++ b/workers/test_env_test.go @@ -0,0 +1,17 @@ +package workers + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestBuildDirNameIsUniquePerEnvironment(t *testing.T) { + createdAt := time.UnixMilli(123) + first := &TestEnvironment{createdAt: createdAt} + second := &TestEnvironment{createdAt: createdAt} + + require.Equal(t, first.buildDirName(), first.buildDirName()) + require.NotEqual(t, first.buildDirName(), second.buildDirName()) +} From 22158029a714799e31f5bb15b294221a74c036c0 Mon Sep 17 00:00:00 2001 From: Thomas Mullaney Date: Mon, 29 Jun 2026 11:37:57 -0700 Subject: [PATCH 2/5] Add targeted benchmark scenarios --- scenarios/external_event.go | 26 +++++++++++++++ scenarios/failure_retry.go | 26 +++++++++++++++ scenarios/visibility_query.go | 63 +++++++++++++++++++++++++++++++++++ 3 files changed, 115 insertions(+) create mode 100644 scenarios/external_event.go create mode 100644 scenarios/failure_retry.go create mode 100644 scenarios/visibility_query.go diff --git a/scenarios/external_event.go b/scenarios/external_event.go new file mode 100644 index 00000000..a2a022c3 --- /dev/null +++ b/scenarios/external_event.go @@ -0,0 +1,26 @@ +package scenarios + +import ( + "github.com/temporalio/omes/loadgen" + . "github.com/temporalio/omes/loadgen/kitchensink" +) + +func init() { + loadgen.MustRegisterScenario(loadgen.Scenario{ + Description: "Each iteration starts one workflow, sends one signal, waits for the signaled state, and completes.", + ExecutorFn: func() loadgen.Executor { + return loadgen.KitchenSinkExecutor{ + TestInput: &TestInput{ + ClientSequence: ClientActions(NewSignalActionsWithIDs(1)...), + WorkflowInput: &WorkflowInput{ + ExpectedSignalCount: 1, + InitialActions: ListActionSet( + NewAwaitWorkflowStateAction("signal_0", "received"), + NewEmptyReturnResultAction(), + ), + }, + }, + } + }, + }) +} diff --git a/scenarios/failure_retry.go b/scenarios/failure_retry.go new file mode 100644 index 00000000..577fa933 --- /dev/null +++ b/scenarios/failure_retry.go @@ -0,0 +1,26 @@ +package scenarios + +import ( + "time" + + "github.com/temporalio/omes/loadgen" + . "github.com/temporalio/omes/loadgen/kitchensink" +) + +func init() { + loadgen.MustRegisterScenario(loadgen.Scenario{ + Description: "Each iteration starts one workflow with one retryable activity that fails once, retries, and completes.", + ExecutorFn: func() loadgen.Executor { + return loadgen.KitchenSinkExecutor{ + TestInput: &TestInput{ + WorkflowInput: &WorkflowInput{ + InitialActions: ListActionSet( + RetryableErrorActivity(1, RemoteActivityWithRetry(1*time.Second, 2, 500*time.Millisecond, 1.0)), + NewEmptyReturnResultAction(), + ), + }, + }, + } + }, + }) +} diff --git a/scenarios/visibility_query.go b/scenarios/visibility_query.go new file mode 100644 index 00000000..c1c41e81 --- /dev/null +++ b/scenarios/visibility_query.go @@ -0,0 +1,63 @@ +package scenarios + +import ( + "context" + "fmt" + "sync/atomic" + "time" + + "github.com/temporalio/omes/loadgen" + . "github.com/temporalio/omes/loadgen/kitchensink" + "go.temporal.io/api/workflowservice/v1" +) + +type visibilityQueryExecutor struct { + completed atomic.Int64 +} + +func init() { + loadgen.MustRegisterScenario(loadgen.Scenario{ + Description: "Each iteration starts and completes one simple workflow, then verifies visibility count for the run.", + ExecutorFn: func() loadgen.Executor { + return &visibilityQueryExecutor{} + }, + }) +} + +func (v *visibilityQueryExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error { + previousOnCompletion := info.Configuration.OnCompletion + info.Configuration.OnCompletion = func(ctx context.Context, run *loadgen.Run) { + v.completed.Add(1) + if previousOnCompletion != nil { + previousOnCompletion(ctx, run) + } + } + + executor := loadgen.KitchenSinkExecutor{ + TestInput: &TestInput{ + WorkflowInput: &WorkflowInput{ + InitialActions: []*ActionSet{ + NoOpSingleActivityActionSet(), + }, + }, + }, + } + if err := executor.Run(ctx, info); err != nil { + return err + } + completed := v.completed.Load() + if completed == 0 { + return fmt.Errorf("no iterations completed") + } + timeout := info.ScenarioOptionDuration(VisibilityVerificationTimeoutFlag, 3*time.Minute) + return loadgen.MinVisibilityCountEventually( + ctx, + info, + &workflowservice.CountWorkflowExecutionsRequest{ + Namespace: info.Namespace, + Query: fmt.Sprintf("%s='%s'", loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID), + }, + int(completed), + timeout, + ) +} From db62e8771c8a03da30b5653b2809d4cf5af0bc34 Mon Sep 17 00:00:00 2001 From: Thomas Mullaney Date: Mon, 29 Jun 2026 16:36:26 -0700 Subject: [PATCH 3/5] Retry transient visibility count failures --- loadgen/helpers.go | 90 +++++++++++++++++------------- loadgen/helpers_visibility_test.go | 36 ++++++++++++ 2 files changed, 86 insertions(+), 40 deletions(-) create mode 100644 loadgen/helpers_visibility_test.go diff --git a/loadgen/helpers.go b/loadgen/helpers.go index 2069f13e..b9fa76b8 100644 --- a/loadgen/helpers.go +++ b/loadgen/helpers.go @@ -57,57 +57,67 @@ func MinVisibilityCountEventually( request *workflowservice.CountWorkflowExecutionsRequest, minCount int, waitAtMost time.Duration, +) error { + lastPrint := time.Now() + return retryVisibilityCount( + ctx, + int64(minCount), + waitAtMost, + 3*time.Second, + func(countCtx context.Context) (int64, error) { + visibilityCount, err := info.Client.CountWorkflow(countCtx, request) + if err != nil { + return 0, err + } + if time.Since(lastPrint) >= 30*time.Second { + info.Logger.Infof("current visibility count: %d (expected at least: %d)\n", + visibilityCount.Count, minCount) + lastPrint = time.Now() + } + return visibilityCount.Count, nil + }, + ) +} + +func retryVisibilityCount( + ctx context.Context, + minCount int64, + waitAtMost time.Duration, + retryInterval time.Duration, + count func(context.Context) (int64, error), ) error { timeoutCtx, cancel := context.WithTimeout(ctx, waitAtMost) defer cancel() - - countTicker := time.NewTicker(3 * time.Second) - defer countTicker.Stop() - - printTicker := time.NewTicker(30 * time.Second) - defer printTicker.Stop() - - var lastVisibilityCount int64 - done := false - - check := func() error { - visibilityCount, err := info.Client.CountWorkflow(timeoutCtx, request) - if err != nil { - return fmt.Errorf("failed to count workflows in visibility: %w", err) - } - lastVisibilityCount = visibilityCount.Count - if lastVisibilityCount >= int64(minCount) { - done = true + var lastCount int64 + var lastErr error + for { + current, err := count(timeoutCtx) + if err == nil { + lastCount = current + lastErr = nil + if current >= minCount { + return nil + } + } else { + lastErr = err } - return nil - } - - // Initial check before entering the loop. - if err := check(); err != nil { - return err - } - - // Loop until we reach the desired count or timeout. - for !done { + timer := time.NewTimer(retryInterval) select { case <-timeoutCtx.Done(): + timer.Stop() + if lastErr != nil { + return fmt.Errorf( + "expected at least %d workflows in visibility, got %d after waiting %v; last count error: %w", + minCount, lastCount, waitAtMost, lastErr, + ) + } return fmt.Errorf( "expected at least %d workflows in visibility, got %d after waiting %v", - minCount, lastVisibilityCount, waitAtMost, + minCount, lastCount, waitAtMost, ) - - case <-printTicker.C: - info.Logger.Infof("current visibility count: %d (expected at least: %d)\n", - lastVisibilityCount, minCount) - - case <-countTicker.C: - if err := check(); err != nil { - return err - } + case <-timer.C: } } - - return nil } // VerifyNoFailedWorkflows verifies that there are no failed or terminated workflows for the given search attribute. diff --git a/loadgen/helpers_visibility_test.go b/loadgen/helpers_visibility_test.go new file mode 100644 index 00000000..f3688ddd --- /dev/null +++ b/loadgen/helpers_visibility_test.go @@ -0,0 +1,36 @@ +package loadgen + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestRetryVisibilityCountIgnoresTransientErrorsUntilMinimumObserved(t *testing.T) { + attempts := 0 + err := retryVisibilityCount( + context.Background(), + 3, + 100*time.Millisecond, + time.Millisecond, + func(context.Context) (int64, error) { + attempts++ + switch attempts { + case 1: + return 0, errors.New("transient deadline") + case 2: + return 1, nil + default: + return 3, nil + } + }, + ) + + if err != nil { + t.Fatal(err) + } + if attempts != 3 { + t.Fatalf("expected three attempts, got %d", attempts) + } +} From 8e86a6da60d15dc131455256e36f62939c0ef57e Mon Sep 17 00:00:00 2001 From: Thomas Mullaney Date: Thu, 2 Jul 2026 10:15:27 -0700 Subject: [PATCH 4/5] Fix matched OMES stress scenario behavior --- scenarios/ebb_and_flow.go | 37 ++++++++------ scenarios/ebb_and_flow_test.go | 10 ++++ scenarios/throughput_stress.go | 43 +++++++++------- scenarios/throughput_stress_test.go | 12 +++++ scenarios/workflow_with_many_actions.go | 21 ++++++++ scenarios/workflow_with_many_actions_test.go | 54 ++++++++++++++++++++ 6 files changed, 143 insertions(+), 34 deletions(-) create mode 100644 scenarios/workflow_with_many_actions_test.go diff --git a/scenarios/ebb_and_flow.go b/scenarios/ebb_and_flow.go index f3fefe22..a62521c1 100644 --- a/scenarios/ebb_and_flow.go +++ b/scenarios/ebb_and_flow.go @@ -49,6 +49,7 @@ type ebbAndFlowConfig struct { MaxConsecutiveErrors int BacklogLogInterval time.Duration VisibilityVerificationTimeout time.Duration + VerifyVisibility bool SleepActivityConfig *loadgen.SleepActivityConfig } @@ -97,6 +98,7 @@ func (e *ebbAndFlowExecutor) Configure(info loadgen.ScenarioInfo) error { MaxConsecutiveErrors: info.ScenarioOptionInt(MaxConsecutiveErrorsFlag, 10), BacklogLogInterval: info.ScenarioOptionDuration(BacklogLogIntervalFlag, 30*time.Second), VisibilityVerificationTimeout: info.ScenarioOptionDuration(VisibilityVerificationTimeoutFlag, 30*time.Second), + VerifyVisibility: info.ScenarioOptionBool(VerifyVisibilityFlag, true), } config.MinBacklog = int64(info.ScenarioOptionInt(MinBacklogFlag, 0)) @@ -224,23 +226,26 @@ func (e *ebbAndFlowExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) return errors.New("No iterations completed. Either the scenario never ran, or it failed to resume correctly.") } - // Post-scenario: verify reported workflow completion count from Visibility. - if err := loadgen.MinVisibilityCountEventually( - ctx, - e.ScenarioInfo, - &workflowservice.CountWorkflowExecutionsRequest{ - Namespace: e.Namespace, - Query: fmt.Sprintf("%s='%s'", - loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID), - }, - totalCompletedWorkflows, - config.VisibilityVerificationTimeout, - ); err != nil { - return err - } + if config.VerifyVisibility { + // Post-scenario: verify reported workflow completion count from Visibility. + if err := loadgen.MinVisibilityCountEventually( + ctx, + e.ScenarioInfo, + &workflowservice.CountWorkflowExecutionsRequest{ + Namespace: e.Namespace, + Query: fmt.Sprintf("%s='%s'", + loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID), + }, + totalCompletedWorkflows, + config.VisibilityVerificationTimeout, + ); err != nil { + return err + } - // Post-scenario: ensure there are no failed or terminated workflows for this run. - return loadgen.VerifyNoFailedWorkflows(ctx, e.ScenarioInfo, loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID) + // Post-scenario: ensure there are no failed or terminated workflows for this run. + return loadgen.VerifyNoFailedWorkflows(ctx, e.ScenarioInfo, loadgen.OmesExecutionIDSearchAttribute, e.ExecutionID) + } + return nil } // Snapshot returns a snapshot of the current state. diff --git a/scenarios/ebb_and_flow_test.go b/scenarios/ebb_and_flow_test.go index 443925af..e76f8799 100644 --- a/scenarios/ebb_and_flow_test.go +++ b/scenarios/ebb_and_flow_test.go @@ -114,3 +114,13 @@ func TestEbbAndFlow(t *testing.T) { require.NoError(t, err, "Executor should complete successfully") }) } + +func TestEbbAndFlowCanDisableVisibilityVerification(t *testing.T) { + executor := newEbbAndFlowExecutor() + info := loadgen.ScenarioInfo{ + ScenarioOptions: map[string]string{VerifyVisibilityFlag: "false"}, + } + + require.NoError(t, executor.Configure(info)) + require.False(t, executor.config.VerifyVisibility) +} diff --git a/scenarios/throughput_stress.go b/scenarios/throughput_stress.go index c134b933..90fc7928 100644 --- a/scenarios/throughput_stress.go +++ b/scenarios/throughput_stress.go @@ -39,6 +39,9 @@ const ( // IncludeRetryScenariosFlag enables retry/timeout/heartbeat activities in throughput_stress. // Default is false. IncludeRetryScenariosFlag = "include-retry-scenarios" + // VerifyVisibilityFlag controls the post-load visibility count and failed-workflow checks. + // Default is true. + VerifyVisibilityFlag = "verify-visibility" ) type tpsState struct { @@ -64,6 +67,7 @@ type tpsConfig struct { ExecutionID string RngSeed int64 IncludeRetryScenarios bool + VerifyVisibility bool } type tpsExecutor struct { @@ -163,6 +167,7 @@ func (t *tpsExecutor) Configure(info loadgen.ScenarioInfo) error { } config.IncludeRetryScenarios = info.ScenarioOptionBool(IncludeRetryScenariosFlag, false) + config.VerifyVisibility = info.ScenarioOptionBool(VerifyVisibilityFlag, true) t.config = config t.rng = rand.New(rand.NewSource(config.RngSeed)) @@ -290,19 +295,26 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error var tpsErrors []error - // Post-scenario: verify reported workflow completion count from Visibility. - if err := loadgen.MinVisibilityCountEventually( - ctx, - info, - &workflowservice.CountWorkflowExecutionsRequest{ - Namespace: info.Namespace, - Query: fmt.Sprintf("%s='%s'", - loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID), - }, - completedWorkflows, - t.config.VisibilityVerificationTimeout, - ); err != nil { - tpsErrors = append(tpsErrors, err) + if t.config.VerifyVisibility { + // Post-scenario: verify reported workflow completion count from Visibility. + if err := loadgen.MinVisibilityCountEventually( + ctx, + info, + &workflowservice.CountWorkflowExecutionsRequest{ + Namespace: info.Namespace, + Query: fmt.Sprintf("%s='%s'", + loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID), + }, + completedWorkflows, + t.config.VisibilityVerificationTimeout, + ); err != nil { + tpsErrors = append(tpsErrors, err) + } + + // Post-scenario: ensure there are no failed or terminated workflows for this run. + if err := loadgen.VerifyNoFailedWorkflows(ctx, info, loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID); err != nil { + tpsErrors = append(tpsErrors, err) + } } // Post-scenario: check throughput threshold @@ -323,11 +335,6 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error } } - // Post-scenario: ensure there are no failed or terminated workflows for this run. - if err := loadgen.VerifyNoFailedWorkflows(ctx, info, loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID); err != nil { - tpsErrors = append(tpsErrors, err) - } - return errors.Join(tpsErrors...) } diff --git a/scenarios/throughput_stress_test.go b/scenarios/throughput_stress_test.go index 47613010..aaf8bea7 100644 --- a/scenarios/throughput_stress_test.go +++ b/scenarios/throughput_stress_test.go @@ -73,3 +73,15 @@ func TestThroughputStress(t *testing.T) { require.NoError(t, err, "Executor should complete successfully when resuming from end") }) } + +func TestThroughputStressCanDisableVisibilityVerification(t *testing.T) { + executor := newThroughputStressExecutor() + info := loadgen.ScenarioInfo{ + RunID: "run-1", + ExecutionID: "execution-1", + ScenarioOptions: map[string]string{VerifyVisibilityFlag: "false"}, + } + + require.NoError(t, executor.Configure(info)) + require.False(t, executor.config.VerifyVisibility) +} diff --git a/scenarios/workflow_with_many_actions.go b/scenarios/workflow_with_many_actions.go index 2325c18b..7b0df80e 100644 --- a/scenarios/workflow_with_many_actions.go +++ b/scenarios/workflow_with_many_actions.go @@ -2,6 +2,7 @@ package scenarios import ( "context" + "fmt" "go.temporal.io/api/common/v1" "go.temporal.io/sdk/converter" "google.golang.org/protobuf/types/known/durationpb" @@ -85,6 +86,26 @@ func init() { ) return nil }, + UpdateWorkflowOptions: func(_ context.Context, run *loadgen.Run, options *loadgen.KitchenSinkWorkflowOptions) error { + childIndex := 0 + for _, actionSet := range options.Params.WorkflowInput.InitialActions { + for _, action := range actionSet.Actions { + child := action.GetExecChildWorkflow() + if child == nil { + continue + } + child.WorkflowId = fmt.Sprintf( + "%s-%s-%d-child-wf-%d", + run.RunID, + run.ExecutionID, + run.Iteration, + childIndex, + ) + childIndex++ + } + } + return nil + }, } }, }) diff --git a/scenarios/workflow_with_many_actions_test.go b/scenarios/workflow_with_many_actions_test.go new file mode 100644 index 00000000..3e0c4f24 --- /dev/null +++ b/scenarios/workflow_with_many_actions_test.go @@ -0,0 +1,54 @@ +package scenarios + +import ( + "context" + "testing" + + "github.com/golang/protobuf/proto" + "github.com/temporalio/omes/loadgen" + "github.com/temporalio/omes/loadgen/kitchensink" + "go.uber.org/zap" +) + +func TestWorkflowWithManyActionsUsesUniqueChildIDsPerIteration(t *testing.T) { + scenario := loadgen.GetScenario("workflow_with_many_actions") + executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor) + if !ok { + t.Fatal("workflow_with_many_actions does not use KitchenSinkExecutor") + } + info := loadgen.ScenarioInfo{ + RunID: "run-1", + ExecutionID: "execution-1", + ScenarioOptions: map[string]string{"children-per-workflow": "2", "activities-per-workflow": "0"}, + Logger: zap.NewNop().Sugar(), + } + if err := executor.PrepareTestInput(context.Background(), info, executor.TestInput); err != nil { + t.Fatal(err) + } + if executor.UpdateWorkflowOptions == nil { + t.Fatal("workflow_with_many_actions does not update child IDs per iteration") + } + params := proto.Clone(executor.TestInput).(*kitchensink.TestInput) + options := loadgen.KitchenSinkWorkflowOptions{Params: params} + if err := executor.UpdateWorkflowOptions(context.Background(), info.NewRun(7), &options); err != nil { + t.Fatal(err) + } + + var childIDs []string + for _, actionSet := range options.Params.WorkflowInput.InitialActions { + for _, action := range actionSet.Actions { + if child := action.GetExecChildWorkflow(); child != nil { + childIDs = append(childIDs, child.WorkflowId) + } + } + } + want := []string{"run-1-execution-1-7-child-wf-0", "run-1-execution-1-7-child-wf-1"} + if len(childIDs) != len(want) { + t.Fatalf("got child IDs %v, want %v", childIDs, want) + } + for index := range want { + if childIDs[index] != want[index] { + t.Fatalf("got child IDs %v, want %v", childIDs, want) + } + } +} From 3cc8535efd82bc320761b2d26263169d7ae03fcc Mon Sep 17 00:00:00 2001 From: Thomas Mullaney Date: Thu, 2 Jul 2026 18:23:49 -0700 Subject: [PATCH 5/5] Add matched cloud benchmark scenarios --- scenarios/failure_retry.go | 2 +- scenarios/failure_retry_test.go | 26 ++++++++++++ .../matched_fixed_resource_consumption.go | 24 +++++++++++ ...matched_fixed_resource_consumption_test.go | 29 +++++++++++++ scenarios/matched_fuzzer.go | 31 ++++++++++++++ scenarios/matched_fuzzer_test.go | 41 +++++++++++++++++++ scenarios/visibility_query.go | 6 ++- scenarios/visibility_query_test.go | 11 +++++ workers/go/kitchensink/kitchen_sink.go | 15 +++++++ workers/go/worker/worker.go | 1 + 10 files changed, 184 insertions(+), 2 deletions(-) create mode 100644 scenarios/failure_retry_test.go create mode 100644 scenarios/matched_fixed_resource_consumption.go create mode 100644 scenarios/matched_fixed_resource_consumption_test.go create mode 100644 scenarios/matched_fuzzer.go create mode 100644 scenarios/matched_fuzzer_test.go create mode 100644 scenarios/visibility_query_test.go diff --git a/scenarios/failure_retry.go b/scenarios/failure_retry.go index 577fa933..17332908 100644 --- a/scenarios/failure_retry.go +++ b/scenarios/failure_retry.go @@ -15,7 +15,7 @@ func init() { TestInput: &TestInput{ WorkflowInput: &WorkflowInput{ InitialActions: ListActionSet( - RetryableErrorActivity(1, RemoteActivityWithRetry(1*time.Second, 2, 500*time.Millisecond, 1.0)), + RetryableErrorActivity(1, RemoteActivityWithRetry(5*time.Second, 2, 500*time.Millisecond, 1.0)), NewEmptyReturnResultAction(), ), }, diff --git a/scenarios/failure_retry_test.go b/scenarios/failure_retry_test.go new file mode 100644 index 00000000..e1f55069 --- /dev/null +++ b/scenarios/failure_retry_test.go @@ -0,0 +1,26 @@ +package scenarios + +import ( + "testing" + "time" + + "github.com/temporalio/omes/loadgen" +) + +func TestFailureRetryAllowsOneFailureAndOneRetry(t *testing.T) { + scenario := loadgen.GetScenario("failure_retry") + executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor) + if !ok { + t.Fatal("failure_retry does not use KitchenSinkExecutor") + } + activity := executor.TestInput.WorkflowInput.InitialActions[0].Actions[0].GetExecActivity() + if got := activity.StartToCloseTimeout.AsDuration(); got != 5*time.Second { + t.Fatalf("got StartToClose timeout %v, want 5s", got) + } + if got := activity.RetryPolicy.MaximumAttempts; got != 2 { + t.Fatalf("got %d maximum attempts, want 2", got) + } + if got := activity.GetRetryableError().FailAttempts; got != 1 { + t.Fatalf("got %d failed attempts, want 1", got) + } +} diff --git a/scenarios/matched_fixed_resource_consumption.go b/scenarios/matched_fixed_resource_consumption.go new file mode 100644 index 00000000..ea32cfda --- /dev/null +++ b/scenarios/matched_fixed_resource_consumption.go @@ -0,0 +1,24 @@ +package scenarios + +import ( + "github.com/temporalio/omes/loadgen" + . "github.com/temporalio/omes/loadgen/kitchensink" +) + +func init() { + loadgen.MustRegisterScenario(loadgen.Scenario{ + Description: "Each iteration runs one bounded 64-KB, 25,000-write resource activity.", + ExecutorFn: func() loadgen.Executor { + return loadgen.KitchenSinkExecutor{ + TestInput: &TestInput{ + WorkflowInput: &WorkflowInput{ + InitialActions: ListActionSet( + GenericActivity("matched_resource", DefaultRemoteActivity), + NewEmptyReturnResultAction(), + ), + }, + }, + } + }, + }) +} diff --git a/scenarios/matched_fixed_resource_consumption_test.go b/scenarios/matched_fixed_resource_consumption_test.go new file mode 100644 index 00000000..adc28abf --- /dev/null +++ b/scenarios/matched_fixed_resource_consumption_test.go @@ -0,0 +1,29 @@ +package scenarios + +import ( + "testing" + + "github.com/temporalio/omes/loadgen" +) + +func TestMatchedFixedResourceConsumptionUsesOneMatchedActivity(t *testing.T) { + scenario := loadgen.GetScenario("matched_fixed_resource_consumption") + if scenario == nil { + t.Fatal("matched_fixed_resource_consumption is not registered") + } + executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor) + if !ok { + t.Fatal("matched_fixed_resource_consumption does not use KitchenSinkExecutor") + } + actions := executor.TestInput.WorkflowInput.InitialActions + if len(actions) != 1 || len(actions[0].Actions) != 2 { + t.Fatalf("got action sets %#v, want one activity and one return", actions) + } + generic := actions[0].Actions[0].GetExecActivity().GetGeneric() + if generic == nil || generic.Type != "matched_resource" { + t.Fatalf("got generic activity %#v, want matched_resource", generic) + } + if actions[0].Actions[1].GetReturnResult() == nil { + t.Fatal("matched resource workflow does not return a result") + } +} diff --git a/scenarios/matched_fuzzer.go b/scenarios/matched_fuzzer.go new file mode 100644 index 00000000..2868a0d1 --- /dev/null +++ b/scenarios/matched_fuzzer.go @@ -0,0 +1,31 @@ +package scenarios + +import ( + "context" + + "github.com/temporalio/omes/loadgen" + "github.com/temporalio/omes/loadgen/kitchensink" +) + +func init() { + loadgen.MustRegisterScenario(loadgen.Scenario{ + Description: "Each iteration follows a deterministic one-, two-, or three-activity path.", + ExecutorFn: func() loadgen.Executor { + return loadgen.KitchenSinkExecutor{ + TestInput: &kitchensink.TestInput{ + WorkflowInput: &kitchensink.WorkflowInput{}, + }, + UpdateWorkflowOptions: func(_ context.Context, run *loadgen.Run, options *loadgen.KitchenSinkWorkflowOptions) error { + pathLength := run.Iteration%3 + 1 + actionSet := &kitchensink.ActionSet{} + for i := 0; i < pathLength; i++ { + actionSet.Actions = append(actionSet.Actions, kitchensink.NoOpSingleActivityActionSet().Actions[0]) + } + actionSet.Actions = append(actionSet.Actions, kitchensink.NoOpSingleActivityActionSet().Actions[1]) + options.Params.WorkflowInput.InitialActions = []*kitchensink.ActionSet{actionSet} + return nil + }, + } + }, + }) +} diff --git a/scenarios/matched_fuzzer_test.go b/scenarios/matched_fuzzer_test.go new file mode 100644 index 00000000..04e1bbed --- /dev/null +++ b/scenarios/matched_fuzzer_test.go @@ -0,0 +1,41 @@ +package scenarios + +import ( + "context" + "testing" + + "github.com/golang/protobuf/proto" + "github.com/temporalio/omes/loadgen" + "github.com/temporalio/omes/loadgen/kitchensink" + "go.uber.org/zap" +) + +func TestMatchedFuzzerUsesDeterministicOneToThreeActivityPaths(t *testing.T) { + scenario := loadgen.GetScenario("matched_fuzzer") + if scenario == nil { + t.Fatal("matched_fuzzer is not registered") + } + executor, ok := scenario.ExecutorFn().(loadgen.KitchenSinkExecutor) + if !ok || executor.UpdateWorkflowOptions == nil { + t.Fatal("matched_fuzzer does not provide dynamic KitchenSink paths") + } + info := loadgen.ScenarioInfo{Logger: zap.NewNop().Sugar()} + for iteration, want := range []int{1, 2, 3, 1} { + params := proto.Clone(executor.TestInput).(*kitchensink.TestInput) + options := loadgen.KitchenSinkWorkflowOptions{Params: params} + if err := executor.UpdateWorkflowOptions(context.Background(), info.NewRun(iteration), &options); err != nil { + t.Fatal(err) + } + got := 0 + for _, actionSet := range options.Params.WorkflowInput.InitialActions { + for _, action := range actionSet.Actions { + if action.GetExecActivity() != nil { + got++ + } + } + } + if got != want { + t.Fatalf("iteration %d got %d activities, want %d", iteration, got, want) + } + } +} diff --git a/scenarios/visibility_query.go b/scenarios/visibility_query.go index c1c41e81..bbfebc60 100644 --- a/scenarios/visibility_query.go +++ b/scenarios/visibility_query.go @@ -15,6 +15,10 @@ type visibilityQueryExecutor struct { completed atomic.Int64 } +func visibilityWorkflowQuery(runID string) string { + return fmt.Sprintf(`WorkflowId STARTS_WITH %q`, "w-"+runID+"-") +} + func init() { loadgen.MustRegisterScenario(loadgen.Scenario{ Description: "Each iteration starts and completes one simple workflow, then verifies visibility count for the run.", @@ -55,7 +59,7 @@ func (v *visibilityQueryExecutor) Run(ctx context.Context, info loadgen.Scenario info, &workflowservice.CountWorkflowExecutionsRequest{ Namespace: info.Namespace, - Query: fmt.Sprintf("%s='%s'", loadgen.OmesExecutionIDSearchAttribute, info.ExecutionID), + Query: visibilityWorkflowQuery(info.RunID), }, int(completed), timeout, diff --git a/scenarios/visibility_query_test.go b/scenarios/visibility_query_test.go new file mode 100644 index 00000000..a75820f1 --- /dev/null +++ b/scenarios/visibility_query_test.go @@ -0,0 +1,11 @@ +package scenarios + +import "testing" + +func TestVisibilityQueryUsesRunWorkflowIDPrefix(t *testing.T) { + got := visibilityWorkflowQuery("run-1") + want := `WorkflowId STARTS_WITH "w-run-1-"` + if got != want { + t.Fatalf("got query %q, want %q", got, want) + } +} diff --git a/workers/go/kitchensink/kitchen_sink.go b/workers/go/kitchensink/kitchen_sink.go index 0a6630d2..80f65c8a 100644 --- a/workers/go/kitchensink/kitchen_sink.go +++ b/workers/go/kitchensink/kitchen_sink.go @@ -370,6 +370,8 @@ func launchActivity(ctx workflow.Context, act *kitchensink.ExecuteActivityAction } else if heartbeat := act.GetHeartbeat(); heartbeat != nil { actType = "heartbeat" args = append(args, heartbeat) + } else if generic := act.GetGeneric(); generic != nil { + actType = generic.Type } if act.GetIsLocal() != nil { opts := workflow.LocalActivityOptions{ @@ -504,6 +506,19 @@ func Noop(_ context.Context) error { return nil } +func MatchedResource(_ context.Context) error { + buffer := make([]byte, 65_536) + var sink uint64 + for i := 0; i < 25_000; i++ { + sink += uint64(i + 1) + buffer[i%len(buffer)] = byte(sink) + } + if len(buffer) == 0 { + return errors.New("resource buffer was not allocated") + } + return nil +} + func Payload(_ context.Context, inputData []byte, bytesToReturn int32) ([]byte, error) { output := make([]byte, bytesToReturn) //goland:noinspection GoDeprecation -- This is fine. We don't need crypto security. diff --git a/workers/go/worker/worker.go b/workers/go/worker/worker.go index 159b192b..ce4881b2 100644 --- a/workers/go/worker/worker.go +++ b/workers/go/worker/worker.go @@ -100,6 +100,7 @@ func runWorkers(client client.Client, taskQueues []string, options clioptions.Wo }) w.RegisterWorkflowWithOptions(kitchensink.KitchenSinkWorkflow, workflow.RegisterOptions{Name: "kitchenSink"}) w.RegisterActivityWithOptions(kitchensink.Noop, activity.RegisterOptions{Name: "noop"}) + w.RegisterActivityWithOptions(kitchensink.MatchedResource, activity.RegisterOptions{Name: "matched_resource"}) w.RegisterActivityWithOptions(kitchensink.Delay, activity.RegisterOptions{Name: "delay"}) w.RegisterActivityWithOptions(kitchensink.Payload, activity.RegisterOptions{Name: "payload"}) w.RegisterActivityWithOptions(kitchensink.RetryableError, activity.RegisterOptions{Name: "retryable_error"})