Merge branch 'main' into cdf/pinned-across-can

Author: carlydf

common/log/tag/tags.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"time"
 
-	deploymentpb "go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/common/primitives"
@@ -975,11 +974,8 @@ func VersioningBehavior(behavior enumspb.VersioningBehavior) ZapTag {
 	return NewStringerTag("versioning-behavior", behavior)
 }
 
-func Deployment(d *deploymentpb.Deployment) ZapTag {
-	if d != nil {
-		return NewAnyTag("deployment", d.SeriesName+":"+d.BuildId)
-	}
-	return NewAnyTag("deployment", "unversioned")
+func Deployment(d string) ZapTag {
+	return NewAnyTag("deployment", d)
 }
 
 func UserDataVersion(v int64) ZapTag {

common/testing/testvars/test_vars.go

@@ -254,8 +254,13 @@ func (tv *TestVars) ExternalDeploymentVersion() *deploymentpb.WorkerDeploymentVe
 	}
 }
 
+// DeploymentVersionString returns v31 string
 func (tv *TestVars) DeploymentVersionString() string {
-	return worker_versioning.WorkerDeploymentVersionToString(tv.DeploymentVersion())
+	return worker_versioning.WorkerDeploymentVersionToStringV31(tv.DeploymentVersion())
+}
+
+func (tv *TestVars) DeploymentVersionStringV32() string {
+	return worker_versioning.ExternalWorkerDeploymentVersionToString(tv.ExternalDeploymentVersion())
 }
 
 func (tv *TestVars) DeploymentVersionTransition() *workflowpb.DeploymentVersionTransition {
@@ -264,7 +269,7 @@ func (tv *TestVars) DeploymentVersionTransition() *workflowpb.DeploymentVersionT
 	}
 	// DescribeWorkflowExecution populates both fields on read, so we expect to see both fields
 	//nolint:staticcheck // SA1019: worker versioning v0.31
-	ret.Version = worker_versioning.ExternalWorkerDeploymentVersionToString(ret.GetDeploymentVersion())
+	ret.Version = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(ret.GetDeploymentVersion())
 	return ret
 }
 

common/worker_versioning/worker_versioning.go

@@ -38,8 +38,11 @@ const (
 	UnversionedSearchAttribute = buildIdSearchAttributePrefixUnversioned
 	UnversionedVersionId       = "__unversioned__"
 
-	// Prefixes, Delimeters and Keys
-	WorkerDeploymentVersionIdDelimiter         = "."
+	// WorkerDeploymentVersionIdDelimiterV31 will be deleted once we stop supporting v31 version string fields
+	// in external and internal APIs. Until then, both delimiters are banned in deployment name. All
+	// deprecated version string fields in APIs keep using the old delimiter. Workflow SA uses new delimiter.
+	WorkerDeploymentVersionIdDelimiterV31      = "."
+	WorkerDeploymentVersionIdDelimiter         = ":"
 	WorkerDeploymentVersionWorkflowIDPrefix    = "temporal-sys-worker-deployment-version"
 	WorkerDeploymentWorkflowIDPrefix           = "temporal-sys-worker-deployment"
 	WorkerDeploymentVersionWorkflowIDDelimeter = ":"
@@ -153,22 +156,46 @@ func BuildIdIfUsingVersioning(stamp *commonpb.WorkerVersionStamp) string {
 
 // DeploymentFromCapabilities returns the deployment if it is using versioning V3, otherwise nil.
 // It returns the deployment from the `options` if present, otherwise, from `capabilities`,
-func DeploymentFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) *deploymentpb.Deployment {
-	if options.GetWorkerVersioningMode() == enumspb.WORKER_VERSIONING_MODE_VERSIONED &&
-		options.GetDeploymentName() != "" &&
-		options.GetBuildId() != "" {
-		return &deploymentpb.Deployment{
-			SeriesName: options.GetDeploymentName(),
-			BuildId:    options.GetBuildId(),
+func DeploymentFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) (*deploymentpb.Deployment, error) {
+	if options.GetWorkerVersioningMode() == enumspb.WORKER_VERSIONING_MODE_VERSIONED {
+		d := options.GetDeploymentName()
+		b := options.GetBuildId()
+		if d == "" {
+			return nil, serviceerror.NewInvalidArgumentf("versioned worker must have deployment name")
+		}
+		if b == "" {
+			return nil, serviceerror.NewInvalidArgumentf("versioned worker must have build id")
+		}
+		if strings.Contains(d, WorkerDeploymentVersionIdDelimiter) || strings.Contains(d, WorkerDeploymentVersionIdDelimiterV31) {
+			// TODO: allow '.' once we get rid of v31 stuff
+			return nil, serviceerror.NewInvalidArgumentf("deployment name cannot contain '%s' or '%s'", WorkerDeploymentVersionIdDelimiter, WorkerDeploymentVersionIdDelimiterV31)
 		}
+		return &deploymentpb.Deployment{
+			SeriesName: d,
+			BuildId:    b,
+		}, nil
 	}
 	if capabilities.GetUseVersioning() && capabilities.GetDeploymentSeriesName() != "" && capabilities.GetBuildId() != "" {
 		return &deploymentpb.Deployment{
 			SeriesName: capabilities.GetDeploymentSeriesName(),
 			BuildId:    capabilities.GetBuildId(),
-		}
+		}, nil
 	}
-	return nil
+	return nil, nil
+}
+
+func DeploymentNameFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) string {
+	if d := options.GetDeploymentName(); d != "" {
+		return d
+	}
+	return capabilities.GetDeploymentSeriesName()
+}
+
+func BuildIdFromCapabilities(capabilities *commonpb.WorkerVersionCapabilities, options *deploymentpb.WorkerDeploymentOptions) string {
+	if d := options.GetBuildId(); d != "" {
+		return d
+	}
+	return capabilities.GetBuildId()
 }
 
 func DeploymentVersionFromOptions(options *deploymentpb.WorkerDeploymentOptions) *deploymentspb.WorkerDeploymentVersion {
@@ -198,16 +225,6 @@ func DeploymentIfValid(d *deploymentpb.Deployment) *deploymentpb.Deployment {
 	return nil
 }
 
-// DeploymentToString is intended to be used for logs and metrics only. Theoretically, it can map
-// different deployments to the string.
-// DO NOT USE IN SERVER LOGIC.
-func DeploymentToString(deployment *deploymentpb.Deployment) string {
-	if deployment == nil {
-		return "UNVERSIONED"
-	}
-	return deployment.GetSeriesName() + ":" + deployment.GetBuildId()
-}
-
 // MakeDirectiveForWorkflowTask returns a versioning directive based on the following parameters:
 // - inheritedBuildId: build ID inherited from a past/previous wf execution (for Child WF or CaN)
 // - assignedBuildId: the build ID to which the WF is currently assigned (i.e. mutable state's AssginedBuildId)
@@ -394,7 +411,12 @@ func ValidateDeployment(deployment *deploymentpb.Deployment) error {
 		return serviceerror.NewInvalidArgument("deployment cannot be nil")
 	}
 	if deployment.GetSeriesName() == "" {
-		return serviceerror.NewInvalidArgument("deployment series name cannot be empty")
+		return serviceerror.NewInvalidArgument("deployment name cannot be empty")
+	}
+	// TODO: remove '.' restriction once the v31 version strings are completely cleaned from external and internal API
+	if strings.Contains(deployment.GetSeriesName(), WorkerDeploymentVersionIdDelimiterV31) ||
+		strings.Contains(deployment.GetSeriesName(), WorkerDeploymentVersionIdDelimiter) {
+		return serviceerror.NewInvalidArgumentf("deployment name cannot contain '%s' or '%s'", WorkerDeploymentVersionIdDelimiterV31, WorkerDeploymentVersionIdDelimiter)
 	}
 	if deployment.GetBuildId() == "" {
 		return serviceerror.NewInvalidArgument("deployment build ID cannot be empty")
@@ -417,13 +439,13 @@ func ValidateDeploymentVersion(version *deploymentspb.WorkerDeploymentVersion) e
 	return nil
 }
 
-// ValidateDeploymentVersionString returns error if the deployment version is nil or it has empty version
+// ValidateDeploymentVersionStringV31 returns error if the deployment version is nil or it has empty version
 // or deployment name.
-func ValidateDeploymentVersionString(version string) (*deploymentspb.WorkerDeploymentVersion, error) {
+func ValidateDeploymentVersionStringV31(version string) (*deploymentspb.WorkerDeploymentVersion, error) {
 	if version == "" {
 		return nil, serviceerror.NewInvalidArgument("version is required")
 	}
-	v, err := WorkerDeploymentVersionFromString(version)
+	v, err := WorkerDeploymentVersionFromStringV31(version)
 	if err != nil {
 		return nil, serviceerror.NewInvalidArgumentf("invalid version string %q, expected format is \"<deployment_name>.<build_id>\"", version)
 	}
@@ -481,7 +503,7 @@ func ValidateVersioningOverride(override *workflowpb.VersioningOverride) error {
 		if override.GetDeployment() != nil {
 			return ValidateDeployment(override.GetDeployment())
 		} else if override.GetPinnedVersion() != "" {
-			_, err := ValidateDeploymentVersionString(override.GetPinnedVersion())
+			_, err := ValidateDeploymentVersionStringV31(override.GetPinnedVersion())
 			return err
 		} else {
 			return serviceerror.NewInvalidArgument("must provide deployment (deprecated) or pinned version if behavior is 'PINNED'")
@@ -625,13 +647,13 @@ func AddV31VersioningInfoToV32(info *workflowpb.WorkflowExecutionVersioningInfo)
 	//nolint:staticcheck // SA1019: worker versioning v0.31
 	if info.Version == "" && info.DeploymentVersion != nil {
 		//nolint:staticcheck // SA1019: worker versioning v0.31
-		info.Version = ExternalWorkerDeploymentVersionToString(info.DeploymentVersion)
+		info.Version = ExternalWorkerDeploymentVersionToStringV31(info.DeploymentVersion)
 	}
 	if t := info.VersionTransition; t != nil {
 		//nolint:staticcheck // SA1019: worker versioning v0.31
 		if t.Version == "" {
 			//nolint:staticcheck // SA1019: worker versioning v0.31
-			t.Version = ExternalWorkerDeploymentVersionToString(t.DeploymentVersion)
+			t.Version = ExternalWorkerDeploymentVersionToStringV31(t.DeploymentVersion)
 		}
 	}
 	if o := info.VersioningOverride; o != nil {
@@ -644,7 +666,7 @@ func AddV31VersioningInfoToV32(info *workflowpb.WorkflowExecutionVersioningInfo)
 				//nolint:staticcheck // SA1019: worker versioning v0.31
 				o.Behavior = enumspb.VERSIONING_BEHAVIOR_PINNED
 				//nolint:staticcheck // SA1019: worker versioning v0.31
-				o.PinnedVersion = ExternalWorkerDeploymentVersionToString(o.GetPinned().GetVersion())
+				o.PinnedVersion = ExternalWorkerDeploymentVersionToStringV31(o.GetPinned().GetVersion())
 			}
 		}
 	}
@@ -675,7 +697,7 @@ func ConvertOverrideToV32(override *workflowpb.VersioningOverride) *workflowpb.V
 			//nolint:staticcheck // SA1019: worker versioning v0.31
 			if override.GetPinnedVersion() != "" {
 				//nolint:staticcheck // SA1019: worker versioning v0.31
-				ret.GetPinned().Version = ExternalWorkerDeploymentVersionFromString(override.GetPinnedVersion())
+				ret.GetPinned().Version = ExternalWorkerDeploymentVersionFromStringV31(override.GetPinnedVersion())
 			} else {
 				//nolint:staticcheck // SA1019: worker versioning v0.30
 				ret.GetPinned().Version = ExternalWorkerDeploymentVersionFromDeployment(override.GetDeployment())
@@ -688,25 +710,32 @@ func ConvertOverrideToV32(override *workflowpb.VersioningOverride) *workflowpb.V
 	return ret
 }
 
-func WorkerDeploymentVersionToString(v *deploymentspb.WorkerDeploymentVersion) string {
+func WorkerDeploymentVersionToStringV31(v *deploymentspb.WorkerDeploymentVersion) string {
 	if v == nil {
 		return "__unversioned__"
 	}
-	return v.GetDeploymentName() + WorkerDeploymentVersionIdDelimiter + v.GetBuildId()
+	return v.GetDeploymentName() + WorkerDeploymentVersionIdDelimiterV31 + v.GetBuildId()
 }
 
 func ExternalWorkerDeploymentVersionToString(v *deploymentpb.WorkerDeploymentVersion) string {
 	if v == nil {
-		return "__unversioned__"
+		return ""
 	}
 	return v.GetDeploymentName() + WorkerDeploymentVersionIdDelimiter + v.GetBuildId()
 }
 
-func ExternalWorkerDeploymentVersionFromString(s string) *deploymentpb.WorkerDeploymentVersion {
+func ExternalWorkerDeploymentVersionToStringV31(v *deploymentpb.WorkerDeploymentVersion) string {
+	if v == nil {
+		return "__unversioned__"
+	}
+	return v.GetDeploymentName() + WorkerDeploymentVersionIdDelimiterV31 + v.GetBuildId()
+}
+
+func ExternalWorkerDeploymentVersionFromStringV31(s string) *deploymentpb.WorkerDeploymentVersion {
 	if s == "" { // unset ramp is no longer supported in v32, so all empty version strings will be treated as unversioned.
 		s = UnversionedVersionId
 	}
-	v, _ := WorkerDeploymentVersionFromString(s)
+	v, _ := WorkerDeploymentVersionFromStringV31(s)
 	if v == nil {
 		return nil
 	}
@@ -716,13 +745,26 @@ func ExternalWorkerDeploymentVersionFromString(s string) *deploymentpb.WorkerDep
 	}
 }
 
-func WorkerDeploymentVersionFromString(s string) (*deploymentspb.WorkerDeploymentVersion, error) {
+func WorkerDeploymentVersionFromStringV31(s string) (*deploymentspb.WorkerDeploymentVersion, error) {
 	if s == "__unversioned__" {
 		return nil, nil
 	}
-	before, after, found := strings.Cut(s, WorkerDeploymentVersionIdDelimiter)
-	if !found {
-		return nil, fmt.Errorf("expected delimiter %s not found in version string %s", WorkerDeploymentVersionIdDelimiter, s)
+	before, after, found := strings.Cut(s, WorkerDeploymentVersionIdDelimiterV31)
+	// Also try parsing via the v32 delimiter in case user is using an old CLI/SDK but passing new version strings.
+	before32, after32, found32 := strings.Cut(s, WorkerDeploymentVersionIdDelimiter)
+	if !found && !found32 {
+		return nil, fmt.Errorf("expected delimiter '%s' or '%s' not found in version string %s", WorkerDeploymentVersionIdDelimiter, WorkerDeploymentVersionIdDelimiterV31, s)
+	}
+	if found && found32 && len(before32) < len(before) {
+		// choose the values based on the delimiter appeared first to ensure that deployment name does not contain any of the banned delimiters
+		before = before32
+		after = after32
+	}
+	if len(before) == 0 {
+		return nil, fmt.Errorf("deployment name is empty in version string %s", s)
+	}
+	if len(after) == 0 {
+		return nil, fmt.Errorf("build id is empty in version string %s", s)
 	}
 	return &deploymentspb.WorkerDeploymentVersion{
 		DeploymentName: before,
@@ -744,7 +786,7 @@ func GetDeploymentNameFromWorkflowID(workflowID string) string {
 // GenerateVersionWorkflowID is a helper that generates a system accepted
 // workflowID which are used in our Worker Deployment Version workflows
 func GenerateVersionWorkflowID(deploymentName string, buildID string) string {
-	versionString := WorkerDeploymentVersionToString(&deploymentspb.WorkerDeploymentVersion{
+	versionString := ExternalWorkerDeploymentVersionToString(&deploymentpb.WorkerDeploymentVersion{
 		DeploymentName: deploymentName,
 		BuildId:        buildID,
 	})

service/frontend/workflow_handler.go

@@ -3369,7 +3369,7 @@ func (wh *WorkflowHandler) DescribeWorkerDeploymentVersion(ctx context.Context,
 		if request.GetDeploymentVersion() == nil {
 			return nil, serviceerror.NewInvalidArgument("deployment version cannot be empty")
 		}
-		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToString(request.GetDeploymentVersion())
+		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(request.GetDeploymentVersion())
 	}
 
 	info, err := wh.workerDeploymentClient.DescribeVersion(ctx, namespaceEntry, versionStr)
@@ -3378,7 +3378,7 @@ func (wh *WorkflowHandler) DescribeWorkerDeploymentVersion(ctx context.Context,
 	}
 
 	//nolint:staticcheck // SA1019: worker versioning v0.31
-	info.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromString(info.Version)
+	info.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(info.Version)
 	return &workflowservice.DescribeWorkerDeploymentVersionResponse{WorkerDeploymentVersionInfo: info}, nil
 }
 
@@ -3416,7 +3416,7 @@ func (wh *WorkflowHandler) SetWorkerDeploymentCurrentVersion(ctx context.Context
 				BuildId:        request.GetBuildId(),
 			}
 		}
-		versionStr = worker_versioning.WorkerDeploymentVersionToString(v)
+		versionStr = worker_versioning.WorkerDeploymentVersionToStringV31(v)
 	}
 
 	resp, err := wh.workerDeploymentClient.SetCurrentVersion(ctx, namespaceEntry, request.DeploymentName, versionStr, request.Identity, request.IgnoreMissingTaskQueues, request.GetConflictToken())
@@ -3430,7 +3430,7 @@ func (wh *WorkflowHandler) SetWorkerDeploymentCurrentVersion(ctx context.Context
 	return &workflowservice.SetWorkerDeploymentCurrentVersionResponse{
 		ConflictToken:             resp.ConflictToken,
 		PreviousVersion:           resp.PreviousVersion,
-		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromString(resp.PreviousVersion),
+		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(resp.PreviousVersion),
 	}, nil
 }
 
@@ -3467,7 +3467,7 @@ func (wh *WorkflowHandler) SetWorkerDeploymentRampingVersion(ctx context.Context
 
 		// This is a v0.32 user trying to ramp up a version. We don't care what percentage it is.
 		if request.GetBuildId() != "" {
-			versionStr = worker_versioning.WorkerDeploymentVersionToString(&deploymentspb.WorkerDeploymentVersion{
+			versionStr = worker_versioning.WorkerDeploymentVersionToStringV31(&deploymentspb.WorkerDeploymentVersion{
 				DeploymentName: request.GetDeploymentName(),
 				BuildId:        request.GetBuildId(),
 			})
@@ -3490,7 +3490,7 @@ func (wh *WorkflowHandler) SetWorkerDeploymentRampingVersion(ctx context.Context
 		ConflictToken:             resp.ConflictToken,
 		PreviousVersion:           resp.PreviousVersion,
 		PreviousPercentage:        resp.PreviousPercentage,
-		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromString(resp.PreviousVersion),
+		PreviousDeploymentVersion: worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(resp.PreviousVersion),
 	}, nil
 }
 
@@ -3565,7 +3565,7 @@ func (wh *WorkflowHandler) DescribeWorkerDeployment(ctx context.Context, request
 
 	for _, vs := range workerDeploymentInfo.VersionSummaries {
 		//nolint:staticcheck // SA1019: worker versioning v0.31
-		vs.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromString(vs.Version)
+		vs.DeploymentVersion = worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(vs.Version)
 	}
 	return &workflowservice.DescribeWorkerDeploymentResponse{
 		WorkerDeploymentInfo: workerDeploymentInfo,
@@ -3611,7 +3611,7 @@ func (wh *WorkflowHandler) DeleteWorkerDeploymentVersion(ctx context.Context, re
 	//nolint:staticcheck // SA1019: worker versioning v0.31
 	versionStr := request.GetVersion()
 	if request.GetDeploymentVersion() != nil {
-		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToString(request.GetDeploymentVersion())
+		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(request.GetDeploymentVersion())
 	}
 
 	err = wh.workerDeploymentClient.DeleteWorkerDeploymentVersion(ctx, namespaceEntry, versionStr, request.SkipDrainage, request.Identity)
@@ -3644,7 +3644,7 @@ func (wh *WorkflowHandler) UpdateWorkerDeploymentVersionMetadata(ctx context.Con
 	//nolint:staticcheck // SA1019: worker versioning v0.31
 	versionStr := request.GetVersion()
 	if request.GetDeploymentVersion() != nil {
-		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToString(request.GetDeploymentVersion())
+		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToStringV31(request.GetDeploymentVersion())
 	}
 
 	identity := uuid.New()

service/history/api/recordactivitytaskstarted/api.go

@@ -176,7 +176,10 @@ func recordActivityTaskStarted(
 	wfBehavior := mutableState.GetEffectiveVersioningBehavior()
 	wfDeployment := mutableState.GetEffectiveDeployment()
 	//nolint:staticcheck // SA1019 deprecated WorkerVersionCapabilities will clean up later
-	pollerDeployment := worker_versioning.DeploymentFromCapabilities(request.PollRequest.WorkerVersionCapabilities, request.PollRequest.DeploymentOptions)
+	pollerDeployment, err := worker_versioning.DeploymentFromCapabilities(request.PollRequest.WorkerVersionCapabilities, request.PollRequest.DeploymentOptions)
+	if err != nil {
+		return nil, rejectCodeUndefined, err
+	}
 	err = worker_versioning.ValidateTaskVersionDirective(request.GetVersionDirective(), wfBehavior, wfDeployment, request.ScheduledDeployment)
 	if err != nil {
 		return nil, rejectCodeUndefined, err