Add support for Worker Deployment Versions (Go worker) (#338)

## Summary

This PR adds `--deployment-name`, `--deployment-build-id`, and
`--default-versioning-behavior` flags to the Omes worker CLI.

## Context

Omes currently only supports the legacy Build ID based worker versioning
scheme for the Go worker. This PR adds flags for Deployment Versions,
and initializes the worker accordingly.

We continue to support the legacy Build ID based path, but I've added a
comment stating that those flags are deprecated. We can phase out this
path entirely in the near future.

Out of scope: Python / TypeScript / Java / .NET worker wiring

## Files changed

| File | Change |
|---|---|
| `cmd/clioptions/worker.go` | Added `DeploymentName`,
`DeploymentBuildID`, `DefaultVersioningBehavior` fields on
`WorkerOptions` and registered `--deployment-name`,
`--deployment-build-id`, `--default-versioning-behavior` flags.
Clarified `--build-id` description as legacy and mutually exclusive with
`--deployment-name`. |
| `workers/go/worker/worker.go` | Hoisted `worker.Options` construction
out of the per-task-queue goroutine. When `--deployment-name` is set,
builds `worker.DeploymentOptions` from `DeploymentName` +
`DeploymentBuildID` + resolved `DefaultVersioningBehavior`; otherwise
keeps the legacy `BuildID` + `UseBuildIDForVersioning` path (now
commented as deprecated). Added `parseVersioningBehavior` helper
(accepts `pinned`, `auto-upgrade`; defaults to `auto-upgrade` on empty).
Added up-front validation for the three mutually-exclusive /
required-together flag combinations. |

## Test plan

- [x] `go build ./...` passes
- [x] Baseline scenario (no versioning flags) completes 3 iterations in
~2s with embedded server
- [x] Deployment-versioned worker starts cleanly and logs `"Started
Worker" ... "BuildID": "v1"` under `--deployment-name omes-test
--deployment-build-id v1 --default-versioning-behavior auto-upgrade`
- [x] `--build-id` + `--deployment-name` → fatal `--build-id and
--deployment-name are mutually exclusive; use --deployment-build-id with
--deployment-name`
- [x] `--deployment-name` without `--deployment-build-id` → fatal
`--deployment-build-id is required when --deployment-name is set`
- [x] `--deployment-build-id` without `--deployment-name` → fatal
`--deployment-build-id requires --deployment-name`
- [x] `--default-versioning-behavior bogus` → fatal `invalid
--default-versioning-behavior "bogus" (expected pinned or auto-upgrade)`
- [x] Legacy `--build-id` path compiles and reaches the deprecated code
branch (runtime end-to-end depends on task-queue Build-ID assignment
rules, unchanged from prior behavior)

Author: veeral-patel

cmd/clioptions/worker.go

@@ -7,6 +7,9 @@ import (
 // WorkerOptions for setting up worker parameters
 type WorkerOptions struct {
 	BuildID                      string
+	DeploymentName               string
+	DeploymentBuildID            string
+	DefaultVersioningBehavior    string
 	MaxConcurrentActivityPollers int
 	MaxConcurrentWorkflowPollers int
 	ActivityPollerAutoscaleMax   int // overrides MaxConcurrentActivityPollers
@@ -25,7 +28,10 @@ func (m *WorkerOptions) FlagSet() *pflag.FlagSet {
 		return m.fs
 	}
 	m.fs = pflag.NewFlagSet("worker_options", pflag.ExitOnError)
-	m.fs.StringVar(&m.BuildID, "build-id", "", "Build ID")
+	m.fs.StringVar(&m.BuildID, "build-id", "", "DEPRECATED: Build ID for legacy Build-ID-based worker versioning. Temporal Server will soon stop supporting the Rules-Based Versioning APIs that back this flag - use --deployment-name and --deployment-build-id instead. Mutually exclusive with --deployment-name")
+	m.fs.StringVar(&m.DeploymentName, "deployment-name", "", "Worker Deployment name. When set, enables Worker Deployment Versioning and must be combined with --deployment-build-id")
+	m.fs.StringVar(&m.DeploymentBuildID, "deployment-build-id", "", "Build ID within the Worker Deployment. Required when --deployment-name is set")
+	m.fs.StringVar(&m.DefaultVersioningBehavior, "default-versioning-behavior", "", "Default versioning behavior for workflows that don't set one at registration. One of: pinned, auto-upgrade. Defaults to auto-upgrade when --deployment-name is set")
 	m.fs.IntVar(&m.MaxConcurrentActivityPollers, "worker-max-concurrent-activity-pollers", 0, "Max concurrent activity pollers")
 	m.fs.IntVar(&m.MaxConcurrentWorkflowPollers, "worker-max-concurrent-workflow-pollers", 0, "Max concurrent workflow pollers")
 	m.fs.IntVar(&m.MaxConcurrentActivities, "worker-max-concurrent-activities", 0, "Max concurrent activities")

workers/go/worker/worker.go

@@ -68,7 +68,61 @@ func makePollerBehavior(simple, auto int) worker.PollerBehavior {
 	})
 }
 
+func parseVersioningBehavior(s string) (workflow.VersioningBehavior, error) {
+	switch s {
+	case "", "auto-upgrade":
+		return workflow.VersioningBehaviorAutoUpgrade, nil
+	case "pinned":
+		return workflow.VersioningBehaviorPinned, nil
+	default:
+		return 0, fmt.Errorf("invalid --default-versioning-behavior %q (expected pinned or auto-upgrade)", s)
+	}
+}
+
 func runWorkers(client client.Client, taskQueues []string, options clioptions.WorkerOptions) error {
+	workerOpts := worker.Options{
+		MaxConcurrentActivityExecutionSize:     options.MaxConcurrentActivities,
+		MaxConcurrentWorkflowTaskExecutionSize: options.MaxConcurrentWorkflowTasks,
+		ActivityTaskPollerBehavior: makePollerBehavior(
+			options.MaxConcurrentActivityPollers,
+			options.ActivityPollerAutoscaleMax,
+		),
+		WorkflowTaskPollerBehavior: makePollerBehavior(
+			options.MaxConcurrentWorkflowPollers,
+			options.WorkflowPollerAutoscaleMax,
+		),
+		WorkerActivitiesPerSecond: options.WorkerActivitiesPerSecond,
+	}
+	if options.DeploymentName != "" {
+		if options.BuildID != "" {
+			return fmt.Errorf("--build-id and --deployment-name are mutually exclusive; use --deployment-build-id with --deployment-name")
+		}
+		if options.DeploymentBuildID == "" {
+			return fmt.Errorf("--deployment-build-id is required when --deployment-name is set")
+		}
+		behavior, err := parseVersioningBehavior(options.DefaultVersioningBehavior)
+		if err != nil {
+			return err
+		}
+		workerOpts.DeploymentOptions = worker.DeploymentOptions{
+			UseVersioning: true,
+			Version: worker.WorkerDeploymentVersion{
+				DeploymentName: options.DeploymentName,
+				BuildID:        options.DeploymentBuildID,
+			},
+			DefaultVersioningBehavior: behavior,
+		}
+	} else if options.DeploymentBuildID != "" {
+		return fmt.Errorf("--deployment-build-id requires --deployment-name")
+	} else if options.BuildID != "" {
+		// DEPRECATED: BuildID and UseBuildIDForVersioning select the legacy
+		// Rules-Based Versioning APIs, which Temporal Server will soon stop
+		// supporting. New users should use --deployment-name and
+		// --deployment-build-id above, which drive Worker Deployment Versioning.
+		workerOpts.BuildID = options.BuildID
+		workerOpts.UseBuildIDForVersioning = true
+	}
+
 	errCh := make(chan error, len(taskQueues))
 	ebbFlowActivities := ebbandflow.Activities{}
 	clientActivities := kitchensink.ClientActivities{
@@ -84,21 +138,7 @@ func runWorkers(client client.Client, taskQueues []string, options clioptions.Wo
 	for _, taskQueue := range taskQueues {
 		taskQueue := taskQueue
 		go func() {
-			w := worker.New(client, taskQueue, worker.Options{
-				BuildID:                                options.BuildID,
-				UseBuildIDForVersioning:                options.BuildID != "",
-				MaxConcurrentActivityExecutionSize:     options.MaxConcurrentActivities,
-				MaxConcurrentWorkflowTaskExecutionSize: options.MaxConcurrentWorkflowTasks,
-				ActivityTaskPollerBehavior: makePollerBehavior(
-					options.MaxConcurrentActivityPollers,
-					options.ActivityPollerAutoscaleMax,
-				),
-				WorkflowTaskPollerBehavior: makePollerBehavior(
-					options.MaxConcurrentWorkflowPollers,
-					options.WorkflowPollerAutoscaleMax,
-				),
-				WorkerActivitiesPerSecond: options.WorkerActivitiesPerSecond,
-			})
+			w := worker.New(client, taskQueue, workerOpts)
 			w.RegisterWorkflowWithOptions(kitchensink.KitchenSinkWorkflow, workflow.RegisterOptions{Name: "kitchenSink"})
 			w.RegisterActivityWithOptions(kitchensink.Noop, activity.RegisterOptions{Name: "noop"})
 			w.RegisterActivityWithOptions(kitchensink.Delay, activity.RegisterOptions{Name: "delay"})