go.temporal.io/server/service/matching/handler.go
726 LOC · 253 covered · 473 uncovered · 39 ranges · 73 concepts · 15 introducers · 17 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package matching
2
3
import (
4
"context"
5
"sync"
6
"time"
7
8
enumspb "go.temporal.io/api/enums/v1"
9
taskqueuepb "go.temporal.io/api/taskqueue/v1"
10
workerpb "go.temporal.io/api/worker/v1"
11
"go.temporal.io/api/workflowservice/v1"
12
"go.temporal.io/server/api/matchingservice/v1"
13
"go.temporal.io/server/common"
14
"go.temporal.io/server/common/cluster"
15
"go.temporal.io/server/common/headers"
16
"go.temporal.io/server/common/log"
17
"go.temporal.io/server/common/membership"
18
"go.temporal.io/server/common/metrics"
19
"go.temporal.io/server/common/namespace"
20
"go.temporal.io/server/common/persistence"
21
"go.temporal.io/server/common/persistence/serialization"
22
"go.temporal.io/server/common/persistence/visibility/manager"
23
"go.temporal.io/server/common/primitives"
24
"go.temporal.io/server/common/resource"
25
"go.temporal.io/server/common/searchattribute"
26
"go.temporal.io/server/common/testing/testhooks"
27
"go.temporal.io/server/common/tqid"
28
"go.temporal.io/server/service/matching/hooks"
29
"go.temporal.io/server/service/matching/workers"
30
"go.temporal.io/server/service/worker/workerdeployment"
31
"go.uber.org/fx"
32
"google.golang.org/protobuf/proto"
33
"google.golang.org/protobuf/reflect/protoreflect"
34
)
35
36
type (
37
// Handler - gRPC handler interface for matchingservice
38
Handler struct {
39
matchingservice.UnimplementedMatchingServiceServer
40
41
engine Engine
42
config *Config
43
metricsHandler metrics.Handler
44
logger log.Logger
45
startWG sync.WaitGroup
46
throttledLogger log.Logger
47
namespaceRegistry namespace.Registry
48
workersRegistry workers.Registry
49
}
50
51
HandlerParams struct {
52
fx.In
53
54
Config *Config
55
Logger log.Logger
56
ThrottledLogger log.Logger
57
TaskManager persistence.TaskManager
58
FairTaskManager persistence.FairTaskManager
59
HistoryClient resource.HistoryClient
60
MatchingRawClient resource.MatchingRawClient
61
WorkerDeploymentClient workerdeployment.Client
62
HostInfoProvider membership.HostInfoProvider
63
MatchingServiceResolver membership.ServiceResolver
64
MetricsHandler metrics.Handler
65
NamespaceRegistry namespace.Registry
66
ClusterMetadata cluster.Metadata
67
NamespaceReplicationQueue persistence.NamespaceReplicationQueue
68
VisibilityManager manager.VisibilityManager
69
NexusEndpointManager persistence.NexusEndpointManager
70
TestHooks testhooks.TestHooks
71
SearchAttributeProvider searchattribute.Provider
72
SearchAttributeMapperProvider searchattribute.MapperProvider
73
RateLimiter TaskDispatchRateLimiter `optional:"true"`
74
WorkersRegistry workers.Registry
75
Serializer serialization.Serializer
76
TaskHookFactories []hooks.TaskHookFactory `group:"TaskHookFactories"`
77
PartitionScalerFactory PartitionScalerFactory
78
}
79
)
80
81
const (
82
serviceName = "temporal.api.workflowservice.v1.MatchingService"
83
)
84
85
var (
86
_ matchingservice.MatchingServiceServer = (*Handler)(nil)
87
)
88
89
// NewHandler creates a gRPC handler for the matchingservice
90
func NewHandler(
91
params HandlerParams,
93
>
handler := &Handler{
94
>
config: params.Config,
95
>
metricsHandler: params.MetricsHandler,
96
>
logger: params.Logger,
97
>
throttledLogger: params.ThrottledLogger,
98
>
engine: NewEngine(
99
>
params.TaskManager,
100
>
params.FairTaskManager,
101
>
params.HistoryClient,
102
>
params.MatchingRawClient, // Use non retry client inside matching
103
>
params.WorkerDeploymentClient,
104
>
params.Config,
105
>
params.Logger,
106
>
params.ThrottledLogger,
107
>
params.MetricsHandler,
108
>
params.NamespaceRegistry,
109
>
params.HostInfoProvider,
110
>
params.MatchingServiceResolver,
111
>
params.ClusterMetadata,
112
>
params.NamespaceReplicationQueue,
113
>
params.VisibilityManager,
114
>
params.NexusEndpointManager,
115
>
params.TestHooks,
116
>
params.SearchAttributeProvider,
117
>
params.SearchAttributeMapperProvider,
118
>
params.RateLimiter,
119
>
params.Serializer,
120
>
params.TaskHookFactories,
121
>
params.PartitionScalerFactory,
122
>
),
123
>
namespaceRegistry: params.NamespaceRegistry,
124
>
workersRegistry: params.WorkersRegistry,
125
>
}
126
>
127
>
// prevent from serving requests before matching engine is started and ready
128
>
handler.startWG.Add(1)
129
>
130
>
return handler
131
>
}
132
133
// Start starts the handler
135
>
h.engine.Start()
136
>
h.startWG.Done()
137
>
}
138
139
// Stop stops the handler
141
>
h.engine.Stop()
142
>
}
143
144
func (h *Handler) opMetricsHandler(
145
namespaceID string,
146
taskQueue *taskqueuepb.TaskQueue,
147
taskQueueType enumspb.TaskQueueType,
148
operation string,
150
>
nsName := h.namespaceName(namespace.ID(namespaceID))
151
>
partition := tqid.UnsafePartitionFromProto(taskQueue, namespaceID, taskQueueType)
152
>
return metrics.GetPerTaskQueuePartitionIDScope(
153
>
h.metricsHandler.WithTags(metrics.OperationTag(operation)),
154
>
nsName.String(),
155
>
partition,
156
>
h.config.BreakdownMetricsByTaskQueue(nsName.String(), partition.TaskQueue().Name(), partition.TaskType()),
157
>
h.config.BreakdownMetricsByPartition(nsName.String(), partition.TaskQueue().Name(), partition.TaskType()),
158
>
)
159
>
}
160
161
// recordNexusTaskRequest emits the nexus_task_requests metric with namespace,
162
// operation, client_name, and is_internal tags.
163
>
func (h *Handler) recordNexusTaskRequest(ctx context.Context, namespaceID string, taskQueueKind enumspb.TaskQueueKind, operation string) {
tags.go ×3
164
>
nsName := h.namespaceName(namespace.ID(namespaceID))
165
>
clientName, _ := headers.GetClientNameAndVersion(ctx)
166
>
isInternal := primitives.IsInternalTaskQueueKind(taskQueueKind)
167
>
metrics.NexusTaskRequests.With(h.metricsHandler).Record(1,
168
>
metrics.NamespaceTag(nsName.String()),
169
>
metrics.OperationTag(operation),
170
>
metrics.ClientNameTag(clientName),
171
>
metrics.IsInternalTag(isInternal),
172
>
)
173
>
}
174
175
// AddActivityTask - adds an activity task.
176
func (h *Handler) AddActivityTask(
177
ctx context.Context,
178
request *matchingservice.AddActivityTaskRequest,
180
>
defer log.CapturePanic(h.logger, &retError)
181
>
startT := time.Now().UTC()
182
>
opMetrics := h.opMetricsHandler(
183
>
request.GetNamespaceId(),
184
>
request.GetTaskQueue(),
185
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
186
>
metrics.MatchingAddActivityTaskScope,
187
>
)
188
>
189
>
if request.GetForwardInfo() != nil {
190
>
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
191
>
}
192
193
>
assignedBuildId, syncMatch, err := h.engine.AddActivityTask(ctx, request)
request_response.pb.go ×6
194
>
if syncMatch {
195
>
metrics.SyncMatchLatencyPerTaskQueue.With(opMetrics).Record(time.Since(startT))
196
>
}
197
>
return &matchingservice.AddActivityTaskResponse{AssignedBuildId: assignedBuildId}, err
198
}
199
200
// AddWorkflowTask - adds a workflow task.
201
func (h *Handler) AddWorkflowTask(
202
ctx context.Context,
203
request *matchingservice.AddWorkflowTaskRequest,
205
>
defer log.CapturePanic(h.logger, &retError)
206
>
startT := time.Now().UTC()
207
>
opMetrics := h.opMetricsHandler(
208
>
request.GetNamespaceId(),
209
>
request.GetTaskQueue(),
210
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
211
>
metrics.MatchingAddWorkflowTaskScope,
212
>
)
213
>
214
>
if request.GetForwardInfo() != nil {
215
>
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
handler.go ×1
216
>
}
217
219
>
if syncMatch {
221
>
}
222
>
return &matchingservice.AddWorkflowTaskResponse{AssignedBuildId: assignedBuildId}, err
handler.go ×25
223
}
224
225
// PollActivityTaskQueue - long poll for an activity task.
226
func (h *Handler) PollActivityTaskQueue(
227
ctx context.Context,
228
request *matchingservice.PollActivityTaskQueueRequest,
230
>
defer log.CapturePanic(h.logger, &retError)
231
>
opMetrics := h.opMetricsHandler(
232
>
request.GetNamespaceId(),
233
>
request.GetPollRequest().GetTaskQueue(),
234
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
235
>
metrics.MatchingPollActivityTaskQueueScope,
236
>
)
237
>
238
>
if request.GetForwardedSource() != "" {
239
>
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
request_response.pb.go ×6
240
>
}
241
243
>
ctx,
244
>
"PollActivityTaskQueue",
245
>
h.throttledLogger,
246
>
); err != nil {
247
return nil, err
248
}
249
251
}
252
253
// PollWorkflowTaskQueue - long poll for a workflow task.
254
func (h *Handler) PollWorkflowTaskQueue(
255
ctx context.Context,
256
request *matchingservice.PollWorkflowTaskQueueRequest,
257
>
) (_ *matchingservice.PollWorkflowTaskQueueResponseWithRawHistory, retError error) {
service_grpc.pb.go ×20
258
>
defer log.CapturePanic(h.logger, &retError)
259
>
opMetrics := h.opMetricsHandler(
260
>
request.GetNamespaceId(),
261
>
request.GetPollRequest().GetTaskQueue(),
262
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
263
>
metrics.MatchingPollWorkflowTaskQueueScope,
264
>
)
265
>
266
>
if request.GetForwardedSource() != "" {
267
>
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
request_response.pb.go ×6
268
>
}
269
271
>
ctx,
272
>
"PollWorkflowTaskQueue",
273
>
h.throttledLogger,
274
>
); err != nil {
275
return nil, err
276
}
277
279
}
280
281
// QueryWorkflow queries a given workflow synchronously and return the query result.
282
func (h *Handler) QueryWorkflow(
283
ctx context.Context,
284
request *matchingservice.QueryWorkflowRequest,
285
) (_ *matchingservice.QueryWorkflowResponse, retError error) {
286
defer log.CapturePanic(h.logger, &retError)
287
opMetrics := h.opMetricsHandler(
288
request.GetNamespaceId(),
289
request.GetTaskQueue(),
290
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
291
metrics.MatchingQueryWorkflowScope,
292
)
293
294
if request.GetForwardInfo() != nil {
295
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
296
}
297
298
return h.engine.QueryWorkflow(ctx, request)
299
}
300
301
// RespondQueryTaskCompleted responds a query task completed
302
func (h *Handler) RespondQueryTaskCompleted(
303
ctx context.Context,
304
request *matchingservice.RespondQueryTaskCompletedRequest,
305
) (_ *matchingservice.RespondQueryTaskCompletedResponse, retError error) {
306
defer log.CapturePanic(h.logger, &retError)
307
opMetrics := h.opMetricsHandler(
308
request.GetNamespaceId(),
309
request.GetTaskQueue(),
310
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
311
metrics.MatchingRespondQueryTaskCompletedScope,
312
)
313
314
err := h.engine.RespondQueryTaskCompleted(ctx, request, opMetrics)
315
return &matchingservice.RespondQueryTaskCompletedResponse{}, err
316
}
317
318
// CancelOutstandingPoll is used to cancel outstanding pollers
319
func (h *Handler) CancelOutstandingPoll(ctx context.Context,
320
>
request *matchingservice.CancelOutstandingPollRequest) (_ *matchingservice.CancelOutstandingPollResponse, retError error) {
service_grpc.pb.go ×19
321
>
defer log.CapturePanic(h.logger, &retError)
322
>
err := h.engine.CancelOutstandingPoll(ctx, request)
323
>
return &matchingservice.CancelOutstandingPollResponse{}, err
324
>
}
325
326
// CancelOutstandingWorkerPolls cancels all outstanding polls for a given worker instance key.
327
func (h *Handler) CancelOutstandingWorkerPolls(ctx context.Context,
328
request *matchingservice.CancelOutstandingWorkerPollsRequest) (_ *matchingservice.CancelOutstandingWorkerPollsResponse, retError error) {
329
defer log.CapturePanic(h.logger, &retError)
330
return h.engine.CancelOutstandingWorkerPolls(ctx, request)
331
}
332
333
// CancelOutstandingWorkerPollsPartition cancels outstanding polls for a worker on a specific partition.
334
func (h *Handler) CancelOutstandingWorkerPollsPartition(ctx context.Context,
335
request *matchingservice.CancelOutstandingWorkerPollsPartitionRequest) (_ *matchingservice.CancelOutstandingWorkerPollsPartitionResponse, retError error) {
336
defer log.CapturePanic(h.logger, &retError)
337
return h.engine.CancelOutstandingWorkerPollsPartition(ctx, request)
338
}
339
340
// DescribeTaskQueue returns information about the target task queue, right now this API returns the
341
// pollers which polled this task queue in last few minutes. If includeTaskQueueStatus field is true,
342
// it will also return status of task queue's ackManager (readLevel, ackLevel, backlogCountHint and taskIDBlock).
343
func (h *Handler) DescribeTaskQueue(
344
ctx context.Context,
345
request *matchingservice.DescribeTaskQueueRequest,
347
>
defer log.CapturePanic(h.logger, &retError)
348
>
resp, err := h.engine.DescribeTaskQueue(ctx, request)
349
>
if err != nil {
350
return nil, err
351
}
352
353
// TODO: remove after 1.24.0-m3
354
>
if len(resp.DescResponse.Pollers) > 0 || resp.DescResponse.TaskQueueStatus != nil {
workflow_handler.go ×11
355
>
// Expand pollerinfo and task queue status into tags 1 and 2 for old frontend to handle
356
>
// proto incompatibility. This only works without ugly protowire code because
357
>
// workflowservice.DescribeTaskQueueResponse and the previous version of
358
>
// matchingservice.DescribeTaskQueueResponse have the same first two fields.
359
>
oldResp := &workflowservice.DescribeTaskQueueResponse{
360
>
Pollers: resp.DescResponse.Pollers,
361
>
TaskQueueStatus: resp.DescResponse.TaskQueueStatus,
362
>
}
363
>
if b, err := proto.Marshal(oldResp); err == nil {
364
>
resp.ProtoReflect().SetUnknown(protoreflect.RawFields(b))
365
>
}
366
}
367
369
}
370
371
func (h *Handler) DescribeVersionedTaskQueues(
372
ctx context.Context,
373
request *matchingservice.DescribeVersionedTaskQueuesRequest,
374
) (_ *matchingservice.DescribeVersionedTaskQueuesResponse, retError error) {
375
defer log.CapturePanic(h.logger, &retError)
376
return h.engine.DescribeVersionedTaskQueues(ctx, request)
377
}
378
379
// DescribeTaskQueuePartition returns information about the target task queue partition.
380
func (h *Handler) DescribeTaskQueuePartition(
381
ctx context.Context,
382
request *matchingservice.DescribeTaskQueuePartitionRequest,
383
) (_ *matchingservice.DescribeTaskQueuePartitionResponse, retError error) {
384
defer log.CapturePanic(h.logger, &retError)
385
return h.engine.DescribeTaskQueuePartition(ctx, request)
386
}
387
388
// ListTaskQueuePartitions returns information about partitions for a taskQueue
389
func (h *Handler) ListTaskQueuePartitions(
390
ctx context.Context,
391
request *matchingservice.ListTaskQueuePartitionsRequest,
392
) (_ *matchingservice.ListTaskQueuePartitionsResponse, retError error) {
393
defer log.CapturePanic(h.logger, &retError)
394
return h.engine.ListTaskQueuePartitions(ctx, request)
395
}
396
397
// UpdateWorkerVersioningRules allows updating the Build ID assignment and redirect rules for a given Task Queue.
398
func (h *Handler) UpdateWorkerVersioningRules(
399
ctx context.Context,
400
request *matchingservice.UpdateWorkerVersioningRulesRequest,
401
) (_ *matchingservice.UpdateWorkerVersioningRulesResponse, retError error) {
402
defer log.CapturePanic(h.logger, &retError)
403
return h.engine.UpdateWorkerVersioningRules(ctx, request)
404
}
405
406
// GetWorkerVersioningRules fetches the Build ID assignment and redirect rules for a Task Queue
407
func (h *Handler) GetWorkerVersioningRules(
408
ctx context.Context,
409
request *matchingservice.GetWorkerVersioningRulesRequest,
410
) (_ *matchingservice.GetWorkerVersioningRulesResponse, retError error) {
411
defer log.CapturePanic(h.logger, &retError)
412
return h.engine.GetWorkerVersioningRules(ctx, request)
413
}
414
415
// UpdateWorkerBuildIdCompatibility allows changing the worker versioning graph for a task queue
416
func (h *Handler) UpdateWorkerBuildIdCompatibility(
417
ctx context.Context,
418
request *matchingservice.UpdateWorkerBuildIdCompatibilityRequest,
419
) (_ *matchingservice.UpdateWorkerBuildIdCompatibilityResponse, retError error) {
420
defer log.CapturePanic(h.logger, &retError)
421
return h.engine.UpdateWorkerBuildIdCompatibility(ctx, request)
422
}
423
424
// GetWorkerBuildIdCompatibility fetches the worker versioning data for a task queue
425
func (h *Handler) GetWorkerBuildIdCompatibility(
426
ctx context.Context,
427
request *matchingservice.GetWorkerBuildIdCompatibilityRequest,
428
) (_ *matchingservice.GetWorkerBuildIdCompatibilityResponse, retError error) {
429
defer log.CapturePanic(h.logger, &retError)
430
return h.engine.GetWorkerBuildIdCompatibility(ctx, request)
431
}
432
433
func (h *Handler) GetTaskQueueUserData(
434
ctx context.Context,
435
request *matchingservice.GetTaskQueueUserDataRequest,
437
>
defer log.CapturePanic(h.logger, &retError)
438
>
return h.engine.GetTaskQueueUserData(ctx, request)
439
>
}
440
441
func (h *Handler) SyncDeploymentUserData(
442
ctx context.Context,
443
request *matchingservice.SyncDeploymentUserDataRequest,
444
) (_ *matchingservice.SyncDeploymentUserDataResponse, retError error) {
445
defer log.CapturePanic(h.logger, &retError)
446
return h.engine.SyncDeploymentUserData(ctx, request)
447
}
448
449
func (h *Handler) ApplyTaskQueueUserDataReplicationEvent(
450
ctx context.Context,
451
request *matchingservice.ApplyTaskQueueUserDataReplicationEventRequest,
452
) (_ *matchingservice.ApplyTaskQueueUserDataReplicationEventResponse, retError error) {
453
defer log.CapturePanic(h.logger, &retError)
454
return h.engine.ApplyTaskQueueUserDataReplicationEvent(ctx, request)
455
}
456
457
func (h *Handler) GetBuildIdTaskQueueMapping(
458
ctx context.Context,
459
request *matchingservice.GetBuildIdTaskQueueMappingRequest,
460
) (_ *matchingservice.GetBuildIdTaskQueueMappingResponse, retError error) {
461
defer log.CapturePanic(h.logger, &retError)
462
return h.engine.GetBuildIdTaskQueueMapping(ctx, request)
463
}
464
465
func (h *Handler) ForceUnloadTaskQueue(
466
ctx context.Context,
467
request *matchingservice.ForceUnloadTaskQueueRequest,
468
) (_ *matchingservice.ForceUnloadTaskQueueResponse, retError error) {
469
defer log.CapturePanic(h.logger, &retError)
470
return h.engine.ForceUnloadTaskQueue(ctx, request)
471
}
472
473
func (h *Handler) ForceUnloadTaskQueuePartition(
474
ctx context.Context,
475
request *matchingservice.ForceUnloadTaskQueuePartitionRequest,
476
>
) (_ *matchingservice.ForceUnloadTaskQueuePartitionResponse, retError error) {
service_grpc.pb.go ×19
477
>
defer log.CapturePanic(h.logger, &retError)
478
>
return h.engine.ForceUnloadTaskQueuePartition(ctx, request)
479
>
}
480
481
func (h *Handler) ForceLoadTaskQueuePartition(
482
ctx context.Context,
483
request *matchingservice.ForceLoadTaskQueuePartitionRequest,
484
>
) (_ *matchingservice.ForceLoadTaskQueuePartitionResponse, retError error) {
request_response.pb.go ×6
485
>
defer log.CapturePanic(h.logger, &retError)
486
>
return h.engine.ForceLoadTaskQueuePartition(ctx, request)
487
>
}
488
489
func (h *Handler) UpdateTaskQueueUserData(
490
ctx context.Context,
491
request *matchingservice.UpdateTaskQueueUserDataRequest,
492
) (_ *matchingservice.UpdateTaskQueueUserDataResponse, retError error) {
493
defer log.CapturePanic(h.logger, &retError)
494
return h.engine.UpdateTaskQueueUserData(ctx, request)
495
}
496
497
func (h *Handler) ReplicateTaskQueueUserData(
498
ctx context.Context,
499
request *matchingservice.ReplicateTaskQueueUserDataRequest,
500
) (_ *matchingservice.ReplicateTaskQueueUserDataResponse, retError error) {
501
defer log.CapturePanic(h.logger, &retError)
502
return h.engine.ReplicateTaskQueueUserData(ctx, request)
503
}
504
505
func (h *Handler) CheckTaskQueueUserDataPropagation(
506
ctx context.Context,
507
request *matchingservice.CheckTaskQueueUserDataPropagationRequest,
508
) (_ *matchingservice.CheckTaskQueueUserDataPropagationResponse, retError error) {
509
defer log.CapturePanic(h.logger, &retError)
510
return h.engine.CheckTaskQueueUserDataPropagation(ctx, request)
511
}
512
513
func (h *Handler) CheckTaskQueueVersionMembership(
514
ctx context.Context,
515
request *matchingservice.CheckTaskQueueVersionMembershipRequest,
516
) (_ *matchingservice.CheckTaskQueueVersionMembershipResponse, retError error) {
517
defer log.CapturePanic(h.logger, &retError)
518
return h.engine.CheckTaskQueueVersionMembership(ctx, request)
519
}
520
521
func (h *Handler) DispatchNexusTask(ctx context.Context, request *matchingservice.DispatchNexusTaskRequest) (_ *matchingservice.DispatchNexusTaskResponse, retError error) {
522
defer log.CapturePanic(h.logger, &retError)
523
return h.engine.DispatchNexusTask(ctx, request)
524
}
525
526
>
func (h *Handler) PollNexusTaskQueue(ctx context.Context, request *matchingservice.PollNexusTaskQueueRequest) (_ *matchingservice.PollNexusTaskQueueResponse, retError error) {
handler.go ×4
527
>
defer log.CapturePanic(h.logger, &retError)
528
>
opMetrics := h.opMetricsHandler(
529
>
request.GetNamespaceId(),
530
>
request.GetRequest().GetTaskQueue(),
531
>
enumspb.TASK_QUEUE_TYPE_NEXUS,
532
>
metrics.MatchingPollWorkflowTaskQueueScope,
533
>
)
534
>
// Only record on the initial handler call (ForwardedSource == ""), not on
535
>
// the forwarded call to the root partition, to avoid double-counting.
536
>
if request.GetForwardedSource() == "" {
537
>
h.recordNexusTaskRequest(ctx, request.GetNamespaceId(), request.GetRequest().GetTaskQueue().GetKind(), "PollNexusTaskQueue")
538
>
}
539
541
h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId()))
542
}
543
545
>
ctx,
546
>
"PollNexusTaskQueue",
547
>
h.throttledLogger,
548
>
); err != nil {
549
return nil, err
550
}
552
}
553
554
>
func (h *Handler) RespondNexusTaskCompleted(ctx context.Context, request *matchingservice.RespondNexusTaskCompletedRequest) (_ *matchingservice.RespondNexusTaskCompletedResponse, retError error) {
request_response.pb.go ×1
555
>
defer log.CapturePanic(h.logger, &retError)
556
>
opMetrics := h.opMetricsHandler(
557
>
request.GetNamespaceId(),
558
>
request.GetTaskQueue(),
559
>
enumspb.TASK_QUEUE_TYPE_NEXUS,
560
>
metrics.MatchingRespondNexusTaskCompletedScope,
561
>
)
562
>
h.recordNexusTaskRequest(ctx, request.GetNamespaceId(), request.GetTaskQueue().GetKind(), "RespondNexusTaskCompleted")
563
>
564
>
return h.engine.RespondNexusTaskCompleted(ctx, request, opMetrics)
565
>
}
566
567
func (h *Handler) RespondNexusTaskFailed(ctx context.Context, request *matchingservice.RespondNexusTaskFailedRequest) (_ *matchingservice.RespondNexusTaskFailedResponse, retError error) {
568
defer log.CapturePanic(h.logger, &retError)
569
opMetrics := h.opMetricsHandler(
570
request.GetNamespaceId(),
571
request.GetTaskQueue(),
572
enumspb.TASK_QUEUE_TYPE_NEXUS,
573
metrics.MatchingRespondNexusTaskFailedScope,
574
)
575
h.recordNexusTaskRequest(ctx, request.GetNamespaceId(), request.GetTaskQueue().GetKind(), "RespondNexusTaskFailed")
576
577
return h.engine.RespondNexusTaskFailed(ctx, request, opMetrics)
578
}
579
580
func (h *Handler) CreateNexusEndpoint(ctx context.Context, request *matchingservice.CreateNexusEndpointRequest) (_ *matchingservice.CreateNexusEndpointResponse, retError error) {
581
defer log.CapturePanic(h.logger, &retError)
582
return h.engine.CreateNexusEndpoint(ctx, request)
583
}
584
585
func (h *Handler) UpdateNexusEndpoint(ctx context.Context, request *matchingservice.UpdateNexusEndpointRequest) (_ *matchingservice.UpdateNexusEndpointResponse, retError error) {
586
defer log.CapturePanic(h.logger, &retError)
587
return h.engine.UpdateNexusEndpoint(ctx, request)
588
}
589
590
func (h *Handler) DeleteNexusEndpoint(ctx context.Context, request *matchingservice.DeleteNexusEndpointRequest) (_ *matchingservice.DeleteNexusEndpointResponse, retError error) {
591
defer log.CapturePanic(h.logger, &retError)
592
return h.engine.DeleteNexusEndpoint(ctx, request)
593
}
594
595
>
func (h *Handler) ListNexusEndpoints(ctx context.Context, request *matchingservice.ListNexusEndpointsRequest) (_ *matchingservice.ListNexusEndpointsResponse, retError error) {
service_grpc.pb.go ×20
596
>
defer log.CapturePanic(h.logger, &retError)
597
>
return h.engine.ListNexusEndpoints(ctx, request)
598
>
}
599
600
// RecordWorkerHeartbeat receive heartbeat request from the worker.
601
func (h *Handler) RecordWorkerHeartbeat(
602
ctx context.Context, request *matchingservice.RecordWorkerHeartbeatRequest,
604
>
defer log.CapturePanic(h.logger, &retError)
605
>
nsID := namespace.ID(request.GetNamespaceId())
606
>
nsName := h.namespaceName(nsID)
607
>
principal := headers.GetPrincipal(ctx)
608
>
609
>
h.workersRegistry.RecordWorkerHeartbeats(nsID, nsName, principal, request.GetHeartbeartRequest().GetWorkerHeartbeat())
610
>
return &matchingservice.RecordWorkerHeartbeatResponse{}, nil
611
>
}
612
613
// ListWorkers retrieves a list of workers in the specified namespace that match the provided filters.
614
func (h *Handler) ListWorkers(
615
_ context.Context, request *matchingservice.ListWorkersRequest,
616
) (_ *matchingservice.ListWorkersResponse, retError error) {
617
defer log.CapturePanic(h.logger, &retError)
618
nsID := namespace.ID(request.GetNamespaceId())
619
listRequest := request.GetListRequest()
620
resp, err := h.workersRegistry.ListWorkers(nsID, workers.ListWorkersParams{
621
Query: listRequest.GetQuery(),
622
PageSize: int(listRequest.GetPageSize()),
623
NextPageToken: listRequest.GetNextPageToken(),
624
IncludeSystemWorkers: listRequest.GetIncludeSystemWorkers(),
625
})
626
if err != nil {
627
return nil, err
628
}
629
// TODO: Stop populating workersInfo once all callers migrate to the Workers field.
630
workersInfo := make([]*workerpb.WorkerInfo, len(resp.Workers))
631
workersList := make([]*workerpb.WorkerListInfo, len(resp.Workers))
632
for i, heartbeat := range resp.Workers {
633
workersInfo[i] = &workerpb.WorkerInfo{
634
WorkerHeartbeat: heartbeat,
635
}
636
workersList[i] = workerHeartbeatToListInfo(heartbeat)
637
}
638
return &matchingservice.ListWorkersResponse{
639
WorkersInfo: workersInfo,
640
Workers: workersList,
641
NextPageToken: resp.NextPageToken,
642
}, nil
643
}
644
645
>
func workerHeartbeatToListInfo(hb *workerpb.WorkerHeartbeat) *workerpb.WorkerListInfo {
handler.go ×1
646
>
hostInfo := hb.GetHostInfo()
647
>
return &workerpb.WorkerListInfo{
648
>
WorkerInstanceKey: hb.GetWorkerInstanceKey(),
649
>
WorkerIdentity: hb.GetWorkerIdentity(),
650
>
TaskQueue: hb.GetTaskQueue(),
651
>
DeploymentVersion: hb.GetDeploymentVersion(),
652
>
SdkName: hb.GetSdkName(),
653
>
SdkVersion: hb.GetSdkVersion(),
654
>
Status: hb.GetStatus(),
655
>
StartTime: hb.GetStartTime(),
656
>
HostName: hostInfo.GetHostName(),
657
>
WorkerGroupingKey: hostInfo.GetWorkerGroupingKey(),
658
>
ProcessId: hostInfo.GetProcessId(),
659
>
Plugins: hb.GetPlugins(),
660
>
Drivers: hb.GetDrivers(),
661
>
}
662
>
}
663
664
func (h *Handler) CountWorkers(
665
_ context.Context, request *matchingservice.CountWorkersRequest,
666
) (_ *matchingservice.CountWorkersResponse, retError error) {
667
defer log.CapturePanic(h.logger, &retError)
668
nsID := namespace.ID(request.GetNamespaceId())
669
countRequest := request.GetCountRequest()
670
count, err := h.workersRegistry.CountWorkers(nsID, countRequest.GetQuery(), countRequest.GetIncludeSystemWorkers())
671
if err != nil {
672
return nil, err
673
}
674
return &matchingservice.CountWorkersResponse{
675
Count: count,
676
}, nil
677
}
678
679
func (h *Handler) UpdateFairnessState(
680
ctx context.Context, request *matchingservice.UpdateFairnessStateRequest,
681
) (_ *matchingservice.UpdateFairnessStateResponse, retError error) {
682
defer log.CapturePanic(h.logger, &retError)
683
return h.engine.UpdateFairnessState(ctx, request)
684
}
685
687
>
entry, err := h.namespaceRegistry.GetNamespaceByID(id)
688
>
if err != nil {
690
>
}
692
}
693
694
>
func (h *Handler) reportForwardedPerTaskQueueCounter(opMetrics metrics.Handler, namespaceId namespace.ID) {
request_response.pb.go ×6
695
>
metrics.ForwardedPerTaskQueueCounter.With(opMetrics).Record(1)
696
>
metrics.MatchingClientForwardedCounter.With(h.metricsHandler).
697
>
Record(
698
>
1,
699
>
metrics.OperationTag(metrics.MatchingAddWorkflowTaskScope),
700
>
metrics.NamespaceTag(h.namespaceName(namespaceId).String()),
701
>
metrics.ServiceRoleTag(metrics.MatchingRoleTagValue))
702
>
}
703
704
func (h *Handler) UpdateTaskQueueConfig(
705
ctx context.Context, request *matchingservice.UpdateTaskQueueConfigRequest,
706
) (_ *matchingservice.UpdateTaskQueueConfigResponse, retError error) {
707
defer log.CapturePanic(h.logger, &retError)
708
return h.engine.UpdateTaskQueueConfig(ctx, request)
709
}
710
711
func (h *Handler) DescribeWorker(
712
_ context.Context, request *matchingservice.DescribeWorkerRequest,
713
) (_ *matchingservice.DescribeWorkerResponse, retError error) {
714
defer log.CapturePanic(h.logger, &retError)
715
nsID := namespace.ID(request.GetNamespaceId())
716
hb, err := h.workersRegistry.DescribeWorker(
717
nsID, request.Request.GetWorkerInstanceKey())
718
if err != nil {
719
return nil, err
720
}
721
return &matchingservice.DescribeWorkerResponse{
722
WorkerInfo: &workerpb.WorkerInfo{
723
WorkerHeartbeat: hb,
724
},
725
}, nil
726
}