Remove scenario name from task queue name (#197)

<!--- Note to EXTERNAL Contributors -->
<!-- Thanks for opening a PR! 
If it is a significant code change, please **make sure there is an open
issue** for this.
We work best with you when we have accepted the idea first before you
code. -->

<!--- For ALL Contributors 👇 -->

## What was changed
<!-- Describe what has changed in this PR -->

Omes client and worker get their task queue _only_ from the RunID now;
instead of also the scenario name.

## Why?
<!-- Tell your future self why have you made these changes -->

RunID is already expected to be a unique identifier; scenario name is
superfluous. When running multiple scenarios, the user can pick a
unique, clear RunID to distinguish them.

This also simplifies the CICD use case where the workers need to be
deployed first before the clients are run. That means the name(s) of the
scenario(s) need to known and propagated ahead of time. Now all that's
needed is a RunID.

## Checklist
<!--- add/delete as needed --->

1. Closes <!-- add issue number here -->

2. How was this tested:
<!--- Please describe how you tested your changes/how we can test them
-->

3. Any docs updates needed?
<!--- update README if applicable
      or point out where to update docs.temporal.io -->

Author: stephanos

README.md

@@ -101,7 +101,7 @@ Notes:
 ### Run a worker for a specific language SDK
 
 ```sh
-go run ./cmd run-worker --scenario workflow_with_single_noop_activity --run-id local-test-run --language go
+go run ./cmd run-worker --run-id local-test-run --language go
 ```
 
 Notes:

cmd/cleanup_scenario.go

@@ -39,17 +39,15 @@ func cleanupScenarioCmd() *cobra.Command {
 
 type scenarioCleaner struct {
 	logger         *zap.SugaredLogger
-	scenario       string
-	runID          string
 	pollInterval   time.Duration
+	scenario       cmdoptions.ScenarioID
 	clientOptions  cmdoptions.ClientOptions
 	metricsOptions cmdoptions.MetricsOptions
 	loggingOptions cmdoptions.LoggingOptions
 }
 
 func (c *scenarioCleaner) addCLIFlags(fs *pflag.FlagSet) {
-	fs.StringVar(&c.scenario, "scenario", "", "Scenario name to cleanup")
-	fs.StringVar(&c.runID, "run-id", "", "Run ID for the run")
+	c.scenario.AddCLIFlags(fs)
 	fs.DurationVar(&c.pollInterval, "poll-interval", time.Second, "Interval for polling completion of job")
 	c.clientOptions.AddCLIFlags(fs)
 	c.metricsOptions.AddCLIFlags(fs, "")
@@ -58,17 +56,17 @@ func (c *scenarioCleaner) addCLIFlags(fs *pflag.FlagSet) {
 
 func (c *scenarioCleaner) run(ctx context.Context) error {
 	c.logger = c.loggingOptions.MustCreateLogger()
-	scenario := loadgen.GetScenario(c.scenario)
+	scenario := loadgen.GetScenario(c.scenario.Scenario)
 	if scenario == nil {
 		return fmt.Errorf("scenario not found")
-	} else if c.runID == "" {
+	} else if c.scenario.RunID == "" {
 		return fmt.Errorf("run ID not found")
 	}
 	metrics := c.metricsOptions.MustCreateMetrics(c.logger)
 	defer metrics.Shutdown(ctx)
 	client := c.clientOptions.MustDial(metrics, c.logger)
 	defer client.Close()
-	taskQueue := loadgen.TaskQueueForRun(c.scenario, c.runID)
+	taskQueue := loadgen.TaskQueueForRun(c.scenario.RunID)
 	jobID := "omes-cleanup-" + taskQueue + "-" + uuid.New().String()
 	username, hostname := "anonymous", "unknown"
 	if user, err := user.Current(); err == nil {

cmd/run_worker.go

@@ -29,7 +29,6 @@ func runWorkerCmd() *cobra.Command {
 		},
 	}
 	r.addCLIFlags(cmd.Flags())
-	cmd.MarkFlagRequired("scenario")
 	cmd.MarkFlagRequired("language")
 	cmd.MarkFlagRequired("run-id")
 	return cmd
@@ -59,7 +58,7 @@ func (r *workerRunner) addCLIFlags(fs *pflag.FlagSet) {
 func (r *workerRunner) preRun() {
 	r.builder.preRun()
 	r.Runner.Builder = r.builder.Builder
-	r.TaskQueueName = loadgen.TaskQueueForRun(r.ScenarioID.Scenario, r.ScenarioID.RunID)
+	r.TaskQueueName = loadgen.TaskQueueForRun(r.ScenarioID.RunID)
 }
 
 func (r *workerRunner) run(ctx context.Context) error {

loadgen/kitchen_sink_executor_test.go

@@ -833,7 +833,7 @@ func testSupportedFeature(
 	}
 	execErr := env.RunExecutorTest(t, testExecutor, scenarioInfo, sdk)
 
-	taskQueueName := TaskQueueForRun(scenarioInfo.ScenarioName, scenarioInfo.RunID)
+	taskQueueName := TaskQueueForRun(scenarioInfo.RunID)
 	historyEvents, historyErr := getWorkflowHistory(t, taskQueueName, env.TemporalClient())
 	if execErr != nil {
 		if len(historyEvents) > 0 {

loadgen/scenario.go

@@ -291,19 +291,19 @@ func (s *ScenarioInfo) RegisterDefaultSearchAttributes(ctx context.Context) erro
 	return nil
 }
 
-// TaskQueueForRun returns a default task queue name for the given scenario name and run ID.
-func TaskQueueForRun(scenarioName, runID string) string {
-	return fmt.Sprintf("%s:%s", scenarioName, runID)
+// TaskQueueForRun returns the task queue name for the given run ID.
+func TaskQueueForRun(runID string) string {
+	return "omes-" + runID
 }
 
 func (r *Run) TaskQueue() string {
-	return TaskQueueForRun(r.ScenarioName, r.RunID)
+	return TaskQueueForRun(r.RunID)
 }
 
 // DefaultStartWorkflowOptions gets default start workflow info.
 func (r *Run) DefaultStartWorkflowOptions() client.StartWorkflowOptions {
 	return client.StartWorkflowOptions{
-		TaskQueue:                                TaskQueueForRun(r.ScenarioName, r.RunID),
+		TaskQueue:                                TaskQueueForRun(r.RunID),
 		ID:                                       fmt.Sprintf("w-%s-%d", r.RunID, r.Iteration),
 		WorkflowExecutionErrorWhenAlreadyStarted: true,
 	}

scenarios/state_transitions_steady.go

@@ -134,7 +134,7 @@ func (s *stateTransitionsSteady) run(ctx context.Context) error {
 		&workflowservice.CountWorkflowExecutionsRequest{
 			Namespace: s.Namespace,
 			Query: fmt.Sprintf("TaskQueue = %q and ExecutionStatus = 'Running'",
-				loadgen.TaskQueueForRun(s.ScenarioName, s.RunID)),
+				loadgen.TaskQueueForRun(s.RunID)),
 		},
 		0,
 		time.Minute,

scenarios/throughput_stress_test.go

@@ -16,7 +16,7 @@ func TestThroughputStress(t *testing.T) {
 
 	scenarioName := "throughput_stress_test"
 	runID := fmt.Sprintf("tps-%d", time.Now().Unix())
-	taskQueueName := loadgen.TaskQueueForRun(scenarioName, runID)
+	taskQueueName := loadgen.TaskQueueForRun(runID)
 
 	env := workers.SetupTestEnvironment(t,
 		workers.WithExecutorTimeout(1*time.Minute),

workers/test_env.go

@@ -178,23 +178,18 @@ func (env *TestEnvironment) RunExecutorTest(
 ) error {
 	env.ensureWorkerBuilt(t, sdk)
 
-	scenarioID := cmdoptions.ScenarioID{
-		Scenario: scenarioInfo.ScenarioName,
-		RunID:    scenarioInfo.RunID,
-	}
-	taskQueueName := loadgen.TaskQueueForRun(scenarioID.Scenario, scenarioID.RunID)
-
 	testCtx, cancelTestCtx := context.WithTimeout(t.Context(), env.executorTimeout)
 	defer cancelTestCtx()
 
-	workerDone := env.startWorker(testCtx, sdk, taskQueueName, scenarioID)
-
 	// Update scenario info with test environment details
 	scenarioInfo.Logger = env.logger.Named("executor")
 	scenarioInfo.MetricsHandler = client.MetricsNopHandler
 	scenarioInfo.Client = env.temporalClient
 	scenarioInfo.Namespace = testNamespace
 
+	taskQueueName := loadgen.TaskQueueForRun(scenarioInfo.RunID)
+	workerDone := env.startWorker(testCtx, sdk, taskQueueName, scenarioInfo)
+
 	err := executor.Run(testCtx, scenarioInfo)
 
 	// Trigger worker shutdown.
@@ -262,7 +257,7 @@ func (env *TestEnvironment) startWorker(
 	ctx context.Context,
 	sdk cmdoptions.Language,
 	taskQueueName string,
-	scenarioID cmdoptions.ScenarioID,
+	scenarioInfo loadgen.ScenarioInfo,
 ) <-chan error {
 	workerDone := make(chan error, 1)
 
@@ -277,7 +272,10 @@ func (env *TestEnvironment) startWorker(
 			},
 			TaskQueueName:            taskQueueName,
 			GracefulShutdownDuration: workerShutdownTimeout,
-			ScenarioID:               scenarioID,
+			ScenarioID: cmdoptions.ScenarioID{
+				Scenario: scenarioInfo.ScenarioName,
+				RunID:    scenarioInfo.RunID,
+			},
 			ClientOptions: cmdoptions.ClientOptions{
 				Address:   env.DevServerAddress(),
 				Namespace: testNamespace,