Implement GetWorkerTaskReachability API #4346

bergundy · 2023-05-16T06:27:28Z

Closes [Worker Versioning] Implement API for UI / CLI to know which workers can be retired #4162
Closes Worker Versioning: Add visibility tags (and possibly API) to enable worker retirement discovery #3303

service/matching/matchingEngine.go

bergundy · 2023-05-16T06:29:14Z

service/frontend/workflow_handler.go

+	response, err := wh.getWorkerTaskReachabilityValidated(ctx, ns, request)
+	if err != nil {
+		wh.logger.Error("Failed getting worker task reachability", tag.Error(err))
+		return nil, serviceerror.NewInternal("Internal error")


I think this is desirable and we should avoid leaking error details to the caller.

bergundy · 2023-05-16T06:30:41Z

service/frontend/task_reachability.go

+			Query:       query.str,
+		}
+
+		// TODO: is count more efficient than select with page size of 1?


I think select with limit of one is more efficient, not sure... @rodrigozhou or @alexshtin could verify.

very possible the count is more efficient, esp. if it can be approximate. databases often keep summary stats to aid the query planner and the count might be able to be answered from those without touching any data at all

bergundy · 2023-05-16T06:31:54Z

service/frontend/task_reachability.go

+)
+
+// Little helper to concurrently map a function over input and fail fast on error.
+func raceMap[IN any, OUT any](input []IN, mapper func(IN) (OUT, error)) ([]OUT, error) {


Not sure if this is idiomatic but I found myself repeating this pattern a few times in this file and ended up refactoring to use this little helper. Maybe we have something similar in the codebase or a shared place we can put this.

I put little generic stuff like this in common/util/util.go

bergundy · 2023-05-16T06:33:08Z

common/persistence/sql/sqlplugin/sqlite/task.go

+	_, err := mdb.conn.ExecContext(ctx, query, params...)
+	if err == nil {
+		// TODO(bergundy)
+		panic("this should be properly tested once we support deletion")


Intentional, I started implementing it but realized I can't test it properly and decided to leave this panic as a reminder that this has to be tested.

I'd say just replace "this" with something slightly more specific

Sure, I didn't think it'd matter much because it's a placeholder for when we do implement deletions.

common/persistence/cassandra/matching_task_store.go

dnr

just a few high-level comments until the new version

cmd/tools/rpcwrappers/main.go

dnr · 2023-05-16T21:53:23Z

service/frontend/task_reachability.go

+)
+
+// Little helper to concurrently map a function over input and fail fast on error.
+func raceMap[IN any, OUT any](input []IN, mapper func(IN) (OUT, error)) ([]OUT, error) {


I put little generic stuff like this in common/util/util.go

dnr · 2023-05-17T22:32:40Z

service/frontend/task_reachability.go

+			Query:       query.str,
+		}
+
+		// TODO: is count more efficient than select with page size of 1?


very possible the count is more efficient, esp. if it can be approximate. databases often keep summary stats to aid the query planner and the count might be able to be answered from those without touching any data at all

…ersioning-build-id-search-attribute

…space

Sushisource

This LGTM but I'm no DB expert

Sushisource · 2023-05-24T21:22:39Z

common/dynamicconfig/constants.go

+	// Do not set this to a value higher than 255 for clusters using SQL based persistence due to predefined VARCHAR
+	// column width.
 	WorkerBuildIdSizeLimit = "limit.workerBuildIdSize"


It'd be good if we just enforced this limit at startup

Hmm.. but this is dynamic config that technically can change at any time.
I do get the point though, I don't know if we support dynamic config validation.
I can check.

Sushisource · 2023-05-24T21:28:27Z

common/persistence/sql/sqlplugin/matching_task_queue.go

@@ -64,6 +64,25 @@ type (
 		DataEncoding  string
 	}

+	AddBuildIdToTaskQueueMapping struct {


Should be plural?

Hmmm... yeah...
My point was that you're adding to the mapping of build id to task queue.
But if it doesn't read right to you, I can change (in a followup PR because I have a couple stacked on top of this).

maybe AddToBuildIdToTaskQueueMapping? that reads okay to me though a little verbose

Sushisource · 2023-05-24T21:28:47Z

common/persistence/sql/sqlplugin/matching_task_queue.go

+		BuildIds      []string
+	}
+
+	RemoveBuildIdToTaskQueueMapping struct {


Also should be plural?

Suggested change

RemoveBuildIdToTaskQueueMapping struct {

RemoveFromBuildIdToTaskQueueMapping struct {

Sushisource · 2023-05-24T21:30:48Z

common/persistence/sql/sqlplugin/sqlite/task.go

+	_, err := mdb.conn.ExecContext(ctx, query, params...)
+	if err == nil {
+		// TODO(bergundy)
+		panic("this should be properly tested once we support deletion")


I'd say just replace "this" with something slightly more specific

Sushisource · 2023-05-24T21:37:24Z

common/util/util.go

+
+// MapConcurrent concurrently maps a function over input and fails fast on error.
+func MapConcurrent[IN any, OUT any](input []IN, mapper func(IN) (OUT, error)) ([]OUT, error) {
+	errorsCh := make(chan error, len(input))


You could have this be only 1 size by using select both where you put in it and where you take out of it, to avoid writing a bunch of nils on non-error cases. Not exactly a huge deal tho.

I don't want to block the goroutines when the receiver exits after the first error.

That's what I mean by using a select on where you take out as well, you can make that nonblocking too.

I would need to use select where I send the results.

Sushisource · 2023-05-24T21:39:07Z

schema/cassandra/temporal/schema.cql

+  build_id_if_row_is_an_index text,      -- If this row is used as a mapping of build id to task queue, this will not be empty
+  data                        blob,      -- temporal.server.api.persistence.v1.TaskQueueUserData
+  data_encoding               text,      -- Encoding type used for serialization, in practice this should always be proto3
+  build_ids                   set<text>, -- All active build ids in all version sets on this task queue (used in an index below)


What does "active" mean? Not deleted? Would be good to clarify

Yes, active is "not deleted", I can clarify.

Sushisource · 2023-05-24T21:52:28Z

service/frontend/workflow_handler.go

+		}
+	}
+	if gotUnversionedRequest && len(request.GetTaskQueues()) == 0 {
+		return nil, serviceerror.NewInvalidArgument("Cannot get reachability of an unversioned worker without specifying at least one task queue")


Probably worth mentioning here that we interpret empty string as unversioned

Sure, couldn't hurt

Sushisource · 2023-05-24T22:01:08Z

tests/advanced_visibility_test.go

+	_, err = s.engine.UpdateWorkerBuildIdCompatibility(ctx, &workflowservice.UpdateWorkerBuildIdCompatibilityRequest{
+		Namespace: s.namespace,
+		TaskQueue: tq1,
+		Operation: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleBuildId{
+			AddNewCompatibleBuildId: &workflowservice.UpdateWorkerBuildIdCompatibilityRequest_AddNewCompatibleVersion{
+				ExistingCompatibleBuildId: v0,
+				NewBuildId:                v01,
+			},
+		},
+	})
+	s.Require().NoError(err)


nit: This test reads slightly more obviously if this update happens after the first one so it's both tq1 ops, then tq2, then tq3

bergundy

Thanks @Sushisource I'll address your comments after I merge my stacked PRs.

…ersioning-redirect-writes

Route task queue user data writes via a single matching node per namespace

Merged David's recent changes Added a reachability tests and search attribute tests Search attribute is now propagated from source version stamp Search attribute has versioned|unversioned prefix to give more accurate reachability results for unversioned workers Merged stacked PR #2

…ersioning-build-id-search-attribute

* Only put messages in the ns replication queue for global namespaces * Add BuildIdBasedVersioning to list of supported capabilities * Address Spencer's comments on #4346

dnr · 2023-05-25T20:37:56Z

cmd/tools/rpcwrappers/main.go

@@ -184,6 +184,17 @@ func makeGetMatchingClient(reqType reflect.Type) string {

 	var tqtPath string
 	switch t.Name() {
+	case "GetBuildIdTaskQueueMappingRequest":
+		// Pick a random node for this request, it's not associated with a specific task queue.
+		tqPath = "&taskqueuepb.TaskQueue{Name: fmt.Sprintf(\"not-applicable-%%s\", rand.Int())}"


it's not being double-formatted, this should work

Suggested change

tqPath = "&taskqueuepb.TaskQueue{Name: fmt.Sprintf(\"not-applicable-%%s\", rand.Int())}"

tqPath = "&taskqueuepb.TaskQueue{Name: fmt.Sprintf(\"not-applicable-%s\", rand.Int())}"

dnr · 2023-05-25T21:01:40Z

common/persistence/cassandra/matching_task_store.go

+	}
+
+	if err := iter.Close(); err != nil {
+		return nil, serviceerror.NewUnavailable(fmt.Sprintf("GetTaskQueuesByBuildId operation failed. Error: %v", err))


we should add a wrapped error field to NewUnavailable since this comes up a lot. not now, though

dnr · 2023-05-25T21:12:50Z

common/persistence/sql/sqlplugin/matching_task_queue.go

@@ -64,6 +64,25 @@ type (
 		DataEncoding  string
 	}

+	AddBuildIdToTaskQueueMapping struct {


maybe AddToBuildIdToTaskQueueMapping? that reads okay to me though a little verbose

dnr · 2023-05-25T21:13:03Z

common/persistence/sql/sqlplugin/matching_task_queue.go

+		BuildIds      []string
+	}
+
+	RemoveBuildIdToTaskQueueMapping struct {


Suggested change

RemoveBuildIdToTaskQueueMapping struct {

RemoveFromBuildIdToTaskQueueMapping struct {

dnr · 2023-05-25T21:13:26Z

common/persistence/sql/sqlplugin/matching_task_queue.go

+		BuildID     string
+	}
+
+	CountTaskQueuesByBuildIdRequest = GetTaskQueuesByBuildIdRequest


make a separate type, not an alias

I will, just curious why

otherwise you could pass a value of one type as the other? I mean, yeah, it's not like it's a big problem, they are the same in some sense, it's just weird to be able to pass a *CountTaskQueuesByBuildIdRequest to GetTaskQueuesByBuildId

dnr · 2023-05-25T22:28:30Z

service/frontend/task_reachability.go

+	statusFilter := ""
+	switch reachabilityType {
+	case enumspb.TASK_REACHABILITY_OPEN_WORKFLOWS:
+		statusFilter = " AND ExecutionStatus = \"Running\""


Suggested change

statusFilter = " AND ExecutionStatus = \"Running\""

statusFilter = ` AND ExecutionStatus = "Running"`

dnr · 2023-05-25T22:29:08Z

service/frontend/task_reachability.go

+	req := manager.CountWorkflowExecutionsRequest{
+		NamespaceID: ns.ID(),
+		Namespace:   ns.Name(),
+		Query:       fmt.Sprintf("TaskQueue = %q AND %s%s", taskQueue, buildIdsFilter, statusFilter),


use constant for search attribute name

dnr · 2023-05-25T22:48:57Z

service/matching/matchingEngine.go

@@ -115,6 +122,10 @@ type (
 		timeSource           clock.TimeSource
 		// Only set if global namespaces are enabled on the cluster.
 		namespaceReplicationQueue persistence.NamespaceReplicationQueue
+		// Disables concurrent task queue user data updates and replication requests (due to a cassandra limitation)
+		namespaceUpdateLockMap map[string]*namespaceUpdateLocks


consider using common/locks.IDMutex here. it takes some of the boilerplate out of this pattern. you can use separate keys for the update + replication locks

I think this is fine TBH

dnr · 2023-05-25T22:54:10Z

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

    // Force unloading a task queue. Used for testing only.
    rpc ForceUnloadTaskQueue (ForceUnloadTaskQueueRequest) returns (ForceUnloadTaskQueueResponse) {}
+
+    // Update task queue user data in owning node for all updates in namespace.


this sounds too tempting to use when you want to update user data. can we make the comment very explicit that these two functions should only be called by the matching implementation itself and that external callers like frontend should use UpdateWorkerBuildIdCompatibility (or some other function for other fields)?

it might even be good if the name starts with Internal

Can change to Internal

dnr · 2023-05-25T22:56:05Z

service/matching/taskQueueManager.go

@@ -488,29 +489,25 @@ func (c *taskQueueManagerImpl) GetUserData(ctx context.Context) (*persistencespb
 }

 //nolint:revive // control coupling


remove this now that lint won't complain?

* Address David's comments * go generate

Note: This commit came from a feature branch and is not expected to build.

Implement GetWorkerTaskReachability API

89988fb

bergundy requested review from dnr and Sushisource May 16, 2023 06:27

bergundy requested a review from a team as a code owner May 16, 2023 06:27

bergundy commented May 16, 2023

View reviewed changes

service/matching/matchingEngine.go Outdated Show resolved Hide resolved

bergundy commented May 16, 2023

View reviewed changes

bergundy added 6 commits May 15, 2023 23:38

Add missing license header

f7a768e

Add missing quota

7b3c6b9

Simplify raceMap

1c0d434

Fix tests using shared versions in namespace scope

7c131a7

Fix frontend quotas unit test

14c3f2d

More test fixes

61280c6

bergundy commented May 16, 2023

View reviewed changes

common/persistence/cassandra/matching_task_store.go Show resolved Hide resolved

bergundy added 2 commits May 16, 2023 16:29

Update API and dependencies

3c230ab

Use new API, add limits for system protection

aa6ae91

bergundy commented May 17, 2023

View reviewed changes

common/persistence/cassandra/matching_task_store.go Outdated Show resolved Hide resolved

dnr reviewed May 17, 2023

View reviewed changes

bergundy added 10 commits May 18, 2023 16:00

Ditch secondary index for build id to task queue mapping

04b1595

Enforce build id to task queue limit on write

7d3523d

Merge remote-tracking branch 'origin/worker-versioning' into worker-v…

40fda4b

…ersioning-build-id-search-attribute

TestMaxTaskQueuesPerBuildIdEnforced

53d6e62

Update protos and dependencies

8c7f6b4

Update to latest feature branch API

50d8717

go mod tidy

dc48add

Merge remote-tracking branch 'origin/worker-versioning' into worker-v…

c3306e7

…ersioning-build-id-search-attribute

Omit deleted build ids from API response

776f695

Oops, accidentally caused replication loop

d50861c

bergundy added 2 commits May 23, 2023 21:28

Test reset + retry

d1bb7ea

Route task queue user data writes via a single matching node per name…

97b8e49

…space

Sushisource approved these changes May 24, 2023

View reviewed changes

bergundy commented May 24, 2023

View reviewed changes

bergundy added 11 commits May 25, 2023 09:07

Merge remote-tracking branch 'origin/worker-versioning' into worker-v…

b986fe8

…ersioning-redirect-writes

Fix tests

04c355c

Merge pull request #2 from bergundy/worker-versioning-redirect-writes

41f6fe3

Route task queue user data writes via a single matching node per namespace

Fix lint?

62fe7bd

Regen protos

becc901

More lint and test fixes

2dba040

Merge remote-tracking branch 'origin/worker-versioning' into worker-v…

11810da

…ersioning-build-id-search-attribute

Fix xdc tests

2f1722f

Fix more xdc tests

186422c

Add missing dynamic config in tests

6993d8a

bergundy merged commit cd407aa into temporalio:worker-versioning May 25, 2023

bergundy deleted the worker-versioning-build-id-search-attribute branch May 25, 2023 20:42

bergundy added a commit to bergundy/temporal that referenced this pull request May 25, 2023

Address Spencer's comments on temporalio#4346

a16d81d

bergundy added a commit to bergundy/temporal that referenced this pull request May 25, 2023

Address Spencer's comments on temporalio#4346

9219418

bergundy added a commit that referenced this pull request May 25, 2023

Worker versioning leftovers (#4400)

bdeb14c

* Only put messages in the ns replication queue for global namespaces * Add BuildIdBasedVersioning to list of supported capabilities * Address Spencer's comments on #4346

dnr reviewed May 25, 2023

View reviewed changes

bergundy added a commit that referenced this pull request May 26, 2023

Address comments from #4346 review (#4405)

ea6246e

* Address David's comments * go generate

dnr pushed a commit that referenced this pull request May 26, 2023

Implement GetWorkerTaskReachability API (#4346)

3075d7c

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit that referenced this pull request May 26, 2023

Address comments from #4346 review (#4405)

b448c08

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit to dnr/temporal that referenced this pull request May 26, 2023

Implement GetWorkerTaskReachability API (temporalio#4346)

a45ba00

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit to dnr/temporal that referenced this pull request May 26, 2023

Address comments from temporalio#4346 review (temporalio#4405)

7618b35

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit that referenced this pull request May 26, 2023

Implement GetWorkerTaskReachability API (#4346)

7af47ac

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit that referenced this pull request May 26, 2023

Address comments from #4346 review (#4405)

0fd893a

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit that referenced this pull request May 26, 2023

Implement GetWorkerTaskReachability API (#4346)

8a325f8

Note: This commit came from a feature branch and is not expected to build.

dnr pushed a commit that referenced this pull request May 26, 2023

Address comments from #4346 review (#4405)

b717268

Note: This commit came from a feature branch and is not expected to build.

	RemoveBuildIdToTaskQueueMapping struct {
	RemoveFromBuildIdToTaskQueueMapping struct {

	tqPath = "&taskqueuepb.TaskQueue{Name: fmt.Sprintf(\"not-applicable-%%s\", rand.Int())}"
	tqPath = "&taskqueuepb.TaskQueue{Name: fmt.Sprintf(\"not-applicable-%s\", rand.Int())}"

	statusFilter = " AND ExecutionStatus = \"Running\""
	statusFilter = ` AND ExecutionStatus = "Running"`

		@@ -488,29 +489,25 @@ func (c taskQueueManagerImpl) GetUserData(ctx context.Context) (persistencespb
		}

		//nolint:revive // control coupling

Implement GetWorkerTaskReachability API #4346

Implement GetWorkerTaskReachability API #4346

Conversation

bergundy commented May 16, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dnr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Sushisource left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bergundy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment