Support ramp in Matching #7126

ShahabT · 2025-01-21T08:27:06Z

What changed?

Matching to Worker Deployment Ramping Version and its percentage when routing tasks. User-facing APIs are not implemented yet.

The implementation also supports ramping from a Deployment Version to unversioned workers.

Why?

Part of Worker Versioning features.

How did you test it?

Added unit and functional tests.

Potential risks

None.

Documentation

None.

Is hotfix candidate?

No

…aking unit-tests

…hab/ramp # Conflicts: # common/testing/testvars/test_vars.go # go.mod # go.sum # service/matching/matching_engine.go

# Conflicts: # client/frontend/client_gen.go # client/frontend/metric_client_gen.go # client/frontend/retryable_client_gen.go # common/dynamicconfig/constants.go # common/rpc/interceptor/logtags/workflow_service_server_gen.go # common/rpc/interceptor/redirection.go # common/testing/mockapi/workflowservicemock/v1/service_grpc.pb.mock.go # common/testing/testvars/test_vars.go # go.mod # go.sum # proto/internal/temporal/server/api/matchingservice/v1/request_response.proto # service/frontend/service.go # service/frontend/workflow_handler.go # service/history/workflow/mutable_state_impl.go # service/history/workflow/mutable_state_impl_test.go # service/history/workflow/util.go # service/history/workflow/workflow_task_state_machine.go # service/matching/config.go # service/matching/physical_task_queue_manager.go # service/worker/workerdeployment/fx.go # tests/deployment_version_test.go

common/worker_versioning/worker_versioning.go

carlydf · 2025-01-22T05:27:08Z

common/worker_versioning/worker_versioning.go

+
+	// Find new current
+	for _, v := range deployments.GetVersions() {
+		// [cleanup-old-wv]


Suggested change

// [cleanup-old-wv]

carlydf · 2025-01-22T05:29:12Z

common/worker_versioning/worker_versioning.go

+
+	// Find ramping version
+	for _, v := range deployments.GetVersions() {
+		// [cleanup-old-wv]


Suggested change

// [cleanup-old-wv]

carlydf · 2025-01-22T05:30:40Z

common/worker_versioning/worker_versioning.go

+		if v.RoutingUpdateTime != nil && v.GetRampPercentage() > 0 {
+			if t := v.RoutingUpdateTime.AsTime(); t.After(ramping.GetRoutingUpdateTime().AsTime()) {
+				ramping = v
+			}
+		}


why not do this in the same for-loop as the current version?

carlydf · 2025-01-22T05:55:35Z

common/worker_versioning/worker_versioning.go

+// FindDeploymentVersionForWorkflowID returns the deployment version that should be used for a
+// particular workflow ID based on the versioning info of the task queue.


Suggested change

// FindDeploymentVersionForWorkflowID returns the deployment version that should be used for a

// particular workflow ID based on the versioning info of the task queue.

// FindDeploymentVersionForWorkflowID returns the deployment version that should be used for a

// particular workflow ID based on the versioning info of the task queue. Nil means unversioned.

carlydf · 2025-01-22T05:58:34Z

proto/internal/temporal/server/api/matchingservice/v1/request_response.proto

+    oneof operation {
+        // The deployment version and its data that is being updated.
+        temporal.server.api.persistence.v1.DeploymentVersionData update_version_data = 6;
+        // Pass to clean up this version's data from the task queue.


Suggested change

// Pass to clean up this version's data from the task queue.

// The version whose data should be cleaned from the task queue.

carlydf · 2025-01-22T06:05:40Z

proto/internal/temporal/server/api/persistence/v1/task_queues.proto

+message DeploymentVersionData {
+    // Special case: `version.version` may be absent (but not `version.deployment_name`). That is
+    // only valid when `ramp_percentage` is non-zero, representing ramping to unversioned in a
+    // particular deployment name.


Suggested change

// particular deployment name.

// particular worker deployment.

carlydf · 2025-01-22T06:09:14Z

proto/internal/temporal/server/api/persistence/v1/task_queues.proto

+    google.protobuf.Timestamp routing_update_time = 2;
+    // If this version is the current version of its deployment.
+    bool is_current = 3;
+    // Range: [0, 100]. Must be zero if is_current is true. Must be non-zero if version is null.


I don't understand what you mean by Must be non-zero if version is null.. It seems like DeploymentVersionData.version == nil is invalid. If we are ramping to unversioned, we still need the deployment name inside version.

carlydf · 2025-01-22T06:34:36Z

service/matching/deployment_util.go

+	return -1
+}
+
+func hasDeploymentVersion(deployments *persistencespb.DeploymentData, deployment *deploymentpb.Deployment) bool {


I think it would be nicer if hasDeploymentVersion took in a dv *deploymentpb.WorkerDeploymentVersion instead of the old deployment *deploymentpb.Deployment but it's not a big deal.

carlydf · 2025-01-22T06:37:13Z

service/matching/task_queue_partition_manager.go

-		// forwarded Query/Nexus task requests do not expire rapidly in contrast to forwarded activity/workflow tasks
-		// that only try up to 200ms sync-match. Therefore, to prevent blocking the request on the wrong build ID, its
-		// more important to allow the parent partition to make a fresh versioning decision in case the child partition
-		// did not have up-to-date User Data when selected a dispatch build ID.


Why do you want to delete this comment? It seems like we're still passing nil forwardInfo

# Conflicts: # common/rpc/interceptor/redirection.go # go.mod # go.sum # service/frontend/workflow_handler.go # tests/deployment_version_test.go

Shivs11 · 2025-01-23T15:28:16Z

common/testing/testvars/test_vars.go

-
-func (tv *TestVars) WithDeploymentNameNumber(n int) *TestVars {
-	return tv.cloneSetN("deployment_name", n)
+func (tv *TestVars) DeploymentVersion() *deploymentpb.WorkerDeploymentVersion {


nit: We should probably rename this to WorkerDeploymentVersion imo - I say this because I almost thought it would return the deployment version (string) rather than the WorkerDeploymentVersion struct it currently returns.

Shivs11 · 2025-01-23T15:46:46Z

proto/internal/temporal/server/api/persistence/v1/task_queues.proto

+    // Special case: `version.version` may be empty (but not `version.deployment_name`). That is
+    // only valid when `ramp_percentage` is non-zero, representing ramping to unversioned in a
+    // particular worker deployment.
+    temporal.api.deployment.v1.WorkerDeploymentVersion version = 1;


quite similar to my previous comment, this is intended to represent the combination of deploymentName + version. IMO, we shouldn't be calling this attribute version since it could just plainly signify the deployment version. I think we should call this worker_deployment_version or any other better name.

Even the comment version.version feels off for the same reason - wdyt

Shivs11 · 2025-01-23T15:49:48Z

proto/internal/temporal/server/api/persistence/v1/task_queues.proto

+    // only valid when `ramp_percentage` is non-zero, representing ramping to unversioned in a
+    // particular worker deployment.


ramp_percentage is non_zero and are ramping to unversioned in a particular worker deployment

the current comment makes it sound that whenever ramp_percentage is zero, we are moving to unversioned which isn't true always

Shivs11 · 2025-01-23T16:07:23Z

common/worker_versioning/worker_versioning.go

+		if v.RoutingUpdateTime != nil && v.GetIsCurrent() {
+			if t := v.RoutingUpdateTime.AsTime(); t.After(current.GetRoutingUpdateTime().AsTime()) {
+				current = v


curious - why do we need a combined check of if v.RoutingUpdateTime != nil && v.GetIsCurrent()? A current deployment version will always have an RoutingUpdateTime right? Or is this just defensive programming to be wary of any potential errors?

Shivs11 · 2025-01-23T16:08:58Z

common/worker_versioning/worker_versioning.go

+	}
+	if ramping.GetRampPercentage() > 0 {
+		info.RampingVersionPercentage = ramping.GetRampPercentage()
+		if ramping.GetVersion().GetVersion() != "" {


stressing on the earlier comment - ramping.GetVersion().GetVersion() looks a bit off to me

Shivs11 · 2025-01-23T16:15:41Z

proto/internal/temporal/server/api/matchingservice/v1/request_response.proto

+        // The deployment version and its data that is being updated.
+        temporal.server.api.persistence.v1.DeploymentVersionData update_version_data = 6;
+        // The version whose data should be cleaned from the task queue.
+        temporal.api.deployment.v1.WorkerDeploymentVersion forget_version = 7;


nit: delete_version?

Shivs11 · 2025-01-23T16:22:03Z

service/history/api/recordactivitytaskstarted/api.go

@@ -272,5 +275,6 @@ func getTaskQueueCurrentDeployment(
 		// The TQ is unversioned
 		return nil, nil
 	}
-	return worker_versioning.FindCurrentDeployment(tqData.GetDeploymentData()), nil
+	versioningInfo := worker_versioning.CalculateTaskQueueVersioningInfo(tqData.GetDeploymentData())
+	return worker_versioning.FindDeploymentVersionForWorkflowID(versioningInfo, workflowId), nil


I like this function's readability - was able to understand everything in one pass - very well written 💡

Shivs11 · 2025-01-23T16:49:44Z

service/matching/task_queue_partition_manager.go

+	if pm.partition.Kind() != enumspb.TASK_QUEUE_KIND_STICKY {
+		perTypeUserData, _, err := pm.getPerTypeUserData()
+		if err != nil {
+			return nil, err
+		}
+		resp.DescResponse.VersioningInfo = worker_versioning.CalculateTaskQueueVersioningInfo(perTypeUserData.GetDeploymentData())
+	}
+	return resp, nil


any specific reason for touching the legacy API? In an idealistic world, isn't the plan to deprecate/reduce usage of the legacy API and encourage folks to move to the Enhanced mode?

Shivs11 · 2025-01-23T16:51:01Z

common/worker_versioning/worker_versioning_test.go

+	v1 = &deploymentpb.WorkerDeploymentVersion{
+		Version:        "v1",
+		DeploymentName: "foo",
+	}
+	v2 = &deploymentpb.WorkerDeploymentVersion{
+		Version:        "v2",
+		DeploymentName: "foo",
+	}
+	v3 = &deploymentpb.WorkerDeploymentVersion{
+		Version:        "v3",
+		DeploymentName: "foo",
+	}


wont block on this but since this is a new file, I think you could have used testVars to get these variables? (with a new added method that returns a struct with a different version but the same deployment name)

Shivs11 · 2025-01-23T17:10:27Z

common/worker_versioning/worker_versioning_test.go

+		{name: "two current + two ramping", want: &taskqueuepb.TaskQueueVersioningInfo{CurrentVersion: v2, UpdateTime: t3, RampingVersion: v3, RampingVersionPercentage: 20},
+			data: &persistencespb.DeploymentData{
+				Versions: []*persistencespb.DeploymentVersionData{
+					{Version: v1, IsCurrent: true, RoutingUpdateTime: t1},
+					{Version: v2, IsCurrent: true, RoutingUpdateTime: t2},
+					{Version: v1, RampPercentage: 50, RoutingUpdateTime: t2},
+					{Version: v3, RampPercentage: 20, RoutingUpdateTime: t3},
+				},
+			},
+		},


confused - can we have two current and two ramping versions in the same deployment at a point in time?

## What changed?  Matching to Worker Deployment Ramping Version and its percentage when routing tasks. User-facing APIs are not implemented yet. The implementation also supports ramping from a Deployment Version to unversioned workers. ## Why?  Part of Worker Versioning features. ## How did you test it?  Added unit and functional tests. ## Potential risks  None. ## Documentation  None. ## Is hotfix candidate?  No --------- Co-authored-by: Shivam Saraf <shivam.saraf@temporal.io>

Shivs11 and others added 10 commits January 16, 2025 15:53

renaming deployment entity wfs

bf85205

fixed cherry-picking conflicts

047bb53

RegisterTaskQueue + DescribeVersion - versioning-3.1

459e119

lint fixes

89a3bf6

Enabling deployment version config for all namespaces

96e12ca

updated proto name according to latest naming conventions + fixed bre…

a205d68

…aking unit-tests

Use new versioning fields

6ff6bbd

update deployment_test

a108af0

Merge remote-tracking branch 'origin/ss/deployment-wf-v-3.1' into sha…

f242caf

…hab/ramp # Conflicts: # common/testing/testvars/test_vars.go # go.mod # go.sum # service/matching/matching_engine.go

Support deployment ramp

84fa1ed

ShahabT requested review from carlydf and Shivs11 January 21, 2025 08:27

ShahabT requested a review from a team as a code owner January 21, 2025 08:27

ShahabT changed the title ~~Shahab/ramp~~ Support ramp in Matching Jan 21, 2025

ShahabT added 2 commits January 21, 2025 15:10

populate DescribeTaskQueue versioning info

0112408

carlydf reviewed Jan 22, 2025

View reviewed changes

ShahabT added 2 commits January 22, 2025 15:56

Merge remote-tracking branch 'origin/versioning-3.1' into shahab/ramp

88bdd4e

# Conflicts: # common/rpc/interceptor/redirection.go # go.mod # go.sum # service/frontend/workflow_handler.go # tests/deployment_version_test.go

address comments

f2141bc

carlydf approved these changes Jan 23, 2025

View reviewed changes

fix test compile

a7e2c19

ShahabT merged commit 36d48a6 into versioning-3.1 Jan 23, 2025
39 of 49 checks passed

ShahabT deleted the shahab/ramp branch January 23, 2025 01:19

Shivs11 reviewed Jan 23, 2025

View reviewed changes

		// FindDeploymentVersionForWorkflowID returns the deployment version that should be used for a
		// particular workflow ID based on the versioning info of the task queue.

	// Pass to clean up this version's data from the task queue.
	// The version whose data should be cleaned from the task queue.

	// particular deployment name.
	// particular worker deployment.

		// only valid when `ramp_percentage` is non-zero, representing ramping to unversioned in a
		// particular worker deployment.

Support ramp in Matching #7126

Support ramp in Matching #7126

Uh oh!

Conversation

ShahabT commented Jan 21, 2025

What changed?

Why?

How did you test it?

Potential risks

Documentation

Is hotfix candidate?

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

carlydf Jan 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

carlydf Jan 22, 2025 •

edited

Loading