go.temporal.io/server/client/matching/client.go
535 LOC · 191 covered · 344 uncovered · 39 ranges · 64 concepts · 8 introducers · 12 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
// Generates all three generated files in this package:
2
//go:generate go run ../../cmd/tools/genrpcwrappers -service matching
3
4
package matching
5
6
import (
7
"context"
8
"fmt"
9
"time"
10
11
"github.com/google/uuid"
12
enumspb "go.temporal.io/api/enums/v1"
13
taskqueuepb "go.temporal.io/api/taskqueue/v1"
14
"go.temporal.io/server/api/matchingservice/v1"
15
"go.temporal.io/server/common"
16
"go.temporal.io/server/common/debug"
17
"go.temporal.io/server/common/dynamicconfig"
18
"go.temporal.io/server/common/goro"
19
"go.temporal.io/server/common/log"
20
"go.temporal.io/server/common/log/tag"
21
"go.temporal.io/server/common/membership"
22
"go.temporal.io/server/common/metrics"
23
"go.temporal.io/server/common/tqid"
24
"google.golang.org/grpc"
25
)
26
27
var _ matchingservice.MatchingServiceClient = (*clientImpl)(nil)
28
29
const (
30
// DefaultTimeout is the max timeout for regular calls
31
DefaultTimeout = time.Minute * debug.TimeoutMultiplier
32
// DefaultLongPollTimeout is the max timeout for long poll calls
33
DefaultLongPollTimeout = time.Minute * 5 * debug.TimeoutMultiplier
34
// evictionCheckInterval is how often departed hosts are reaped from the cache.
35
evictionCheckInterval = 30 * time.Second
36
)
37
38
type clientImpl struct {
39
timeout time.Duration
40
longPollTimeout time.Duration
41
clients common.ClientCache
42
resolver membership.ServiceResolver
43
connectionCloseDelay dynamicconfig.DurationPropertyFn
44
metricsHandler metrics.Handler
45
logger log.Logger
46
loadBalancer LoadBalancer
47
spreadRouting dynamicconfig.TypedPropertyFn[dynamicconfig.GradualChange[int]]
48
partitionCache *partitionCache
49
evictionWatcher *goro.Handle
50
}
51
52
// NewClient creates a new matching service gRPC client
53
func NewClient(
54
timeout time.Duration,
55
longPollTimeout time.Duration,
56
clients common.ClientCache,
57
metricsHandler metrics.Handler,
58
logger log.Logger,
59
lb LoadBalancer,
60
spreadRouting dynamicconfig.TypedPropertyFn[dynamicconfig.GradualChange[int]],
61
resolver membership.ServiceResolver,
62
connectionCloseDelay dynamicconfig.DurationPropertyFn,
64
>
c := &clientImpl{
65
>
timeout: timeout,
66
>
longPollTimeout: longPollTimeout,
67
>
clients: clients,
68
>
resolver: resolver,
69
>
connectionCloseDelay: connectionCloseDelay,
70
>
metricsHandler: metricsHandler,
71
>
logger: logger,
72
>
loadBalancer: lb,
73
>
spreadRouting: spreadRouting,
74
>
partitionCache: newPartitionCache(metricsHandler),
75
>
}
76
>
77
>
// Start goroutine to prune partition count cache. Stopped by Stop().
78
>
c.partitionCache.Start()
79
>
80
>
// Evict cached clients whose host leaves the membership ring. Stopped by Stop().
81
>
c.evictionWatcher = goro.NewHandle(context.Background()).Go(c.watchMembership)
82
>
83
>
return c
84
>
}
85
86
// Stop deterministically releases the resources started by NewClient: it stops
87
// the eviction watcher and partition-cache rotation goroutines and closes every
88
// cached gRPC connection. It is safe to call more than once.
90
>
c.evictionWatcher.Cancel()
91
>
<-c.evictionWatcher.Done()
92
>
c.partitionCache.Stop()
93
>
c.clients.EvictAll()
94
>
}
95
96
// watchMembership evicts cached clients whose host leaves the membership ring.
97
// It runs until ctx is cancelled (by Stop).
99
>
listenerName := fmt.Sprintf("matchingClientCache-%s", uuid.New().String())
100
>
ch := make(chan *membership.ChangedEvent, 1)
101
>
if err := c.resolver.AddListener(listenerName, ch); err != nil {
102
c.logger.Error("Failed to subscribe matching cache to membership", tag.Error(err))
103
return err
104
}
106
107
// Reap departed hosts via a per-address deadline checked by a single ticker;
108
// a re-add resets it to the latest removal.
110
>
ticker := time.NewTicker(evictionCheckInterval)
111
>
defer ticker.Stop()
112
>
for {
113
>
select {
115
>
return nil
117
>
for _, h := range event.HostsRemoved {
119
>
}
121
>
delete(evictAt, h.GetAddress())
122
>
}
123
case <-ticker.C:
124
reapEvictableClients(c.resolver, c.clients, evictAt)
125
}
126
}
127
}
128
129
func reapEvictableClients(
130
resolver membership.ServiceResolver,
131
clients common.ClientCache,
132
evictAt map[string]time.Time,
133
) {
134
if len(evictAt) == 0 {
135
return
136
}
137
members := make(map[string]struct{})
138
for _, m := range resolver.Members() {
139
members[m.GetAddress()] = struct{}{}
140
}
141
now := time.Now()
142
for addr, deadline := range evictAt {
143
if _, ok := members[addr]; ok {
144
delete(evictAt, addr) // back in the ring; cancel the eviction
145
continue
146
}
147
if now.Before(deadline) {
148
continue
149
}
150
clients.Evict(addr)
151
delete(evictAt, addr)
152
}
153
}
154
155
func (c *clientImpl) AddActivityTask(
156
ctx context.Context,
157
request *matchingservice.AddActivityTaskRequest,
158
opts ...grpc.CallOption,
160
>
if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
161
return c.addActivityTask(ctx, PartitionCounts{}, request, opts)
162
}
164
>
request.GetNamespaceId(),
165
>
request.GetTaskQueue().GetName(),
166
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
167
>
)
168
>
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.addActivityTask)
169
}
170
171
func (c *clientImpl) addActivityTask(
172
ctx context.Context,
173
pc PartitionCounts,
174
request *matchingservice.AddActivityTaskRequest,
175
opts []grpc.CallOption,
177
>
request = common.CloneProto(request)
178
>
client, err := c.pickClientForWrite(
179
>
request.GetTaskQueue(),
180
>
request.GetNamespaceId(),
181
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
182
>
request.GetForwardInfo().GetSourcePartition(),
183
>
pc,
184
>
)
185
>
if err != nil {
186
return nil, err
187
}
189
>
defer cancel()
190
>
191
>
return client.AddActivityTask(ctx, request, opts...)
192
}
193
194
func (c *clientImpl) AddWorkflowTask(
195
ctx context.Context,
196
request *matchingservice.AddWorkflowTaskRequest,
198
>
if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
200
>
}
202
>
request.GetNamespaceId(),
203
>
request.GetTaskQueue().GetName(),
204
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
205
>
)
206
>
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.addWorkflowTask)
207
}
208
209
func (c *clientImpl) addWorkflowTask(
210
ctx context.Context,
211
pc PartitionCounts,
212
request *matchingservice.AddWorkflowTaskRequest,
213
opts []grpc.CallOption,
215
>
request = common.CloneProto(request)
216
>
client, err := c.pickClientForWrite(
217
>
request.GetTaskQueue(),
218
>
request.GetNamespaceId(),
219
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
220
>
request.GetForwardInfo().GetSourcePartition(),
221
>
pc,
222
>
)
223
>
if err != nil {
224
return nil, err
225
}
227
>
defer cancel()
228
>
return client.AddWorkflowTask(ctx, request, opts...)
229
}
230
231
func (c *clientImpl) PollActivityTaskQueue(
232
ctx context.Context,
233
request *matchingservice.PollActivityTaskQueueRequest,
234
opts ...grpc.CallOption,
236
>
if !isPartitionAwareKind(request.GetPollRequest().GetTaskQueue().GetKind()) {
237
return c.pollActivityTaskQueue(ctx, PartitionCounts{}, request, opts)
238
}
240
>
request.GetNamespaceId(),
241
>
request.GetPollRequest().GetTaskQueue().GetName(),
242
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
243
>
)
244
>
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.pollActivityTaskQueue)
245
}
246
247
func (c *clientImpl) pollActivityTaskQueue(
248
ctx context.Context,
249
pc PartitionCounts,
250
request *matchingservice.PollActivityTaskQueueRequest,
251
opts []grpc.CallOption,
253
>
request = common.CloneProto(request)
254
>
client, release, err := c.pickClientForRead(
255
>
request.GetPollRequest().GetTaskQueue(),
256
>
request.GetNamespaceId(),
257
>
enumspb.TASK_QUEUE_TYPE_ACTIVITY,
258
>
request.GetForwardedSource(),
259
>
pc,
260
>
)
261
>
if err != nil {
262
return nil, err
263
}
265
>
defer release()
266
>
}
267
>
ctx, cancel := c.createLongPollContext(ctx)
268
>
defer cancel()
269
>
return client.PollActivityTaskQueue(ctx, request, opts...)
270
}
271
272
func (c *clientImpl) PollWorkflowTaskQueue(
273
ctx context.Context,
274
request *matchingservice.PollWorkflowTaskQueueRequest,
275
opts ...grpc.CallOption,
277
>
if !isPartitionAwareKind(request.GetPollRequest().GetTaskQueue().GetKind()) {
278
>
return c.pollWorkflowTaskQueue(ctx, PartitionCounts{}, request, opts)
279
>
}
280
>
pkey := c.partitionCache.makeKey(
281
>
request.GetNamespaceId(),
282
>
request.GetPollRequest().GetTaskQueue().GetName(),
283
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
284
>
)
285
>
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.pollWorkflowTaskQueue)
286
}
287
288
func (c *clientImpl) pollWorkflowTaskQueue(
289
ctx context.Context,
290
pc PartitionCounts,
291
request *matchingservice.PollWorkflowTaskQueueRequest,
292
opts []grpc.CallOption,
294
>
request = common.CloneProto(request)
295
>
client, release, err := c.pickClientForRead(
296
>
request.GetPollRequest().GetTaskQueue(),
297
>
request.GetNamespaceId(),
298
>
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
299
>
request.GetForwardedSource(),
300
>
pc,
301
>
)
302
>
if err != nil {
303
return nil, err
304
}
306
>
defer release()
307
>
}
308
>
ctx, cancel := c.createLongPollContext(ctx)
309
>
defer cancel()
310
>
return client.PollWorkflowTaskQueue(ctx, request, opts...)
311
}
312
313
func (c *clientImpl) QueryWorkflow(
314
ctx context.Context,
315
request *matchingservice.QueryWorkflowRequest,
316
opts ...grpc.CallOption,
317
) (*matchingservice.QueryWorkflowResponse, error) {
318
if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
319
return c.queryWorkflow(ctx, PartitionCounts{}, request, opts)
320
}
321
pkey := c.partitionCache.makeKey(
322
request.GetNamespaceId(),
323
request.GetTaskQueue().GetName(),
324
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
325
)
326
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.queryWorkflow)
327
}
328
329
func (c *clientImpl) queryWorkflow(
330
ctx context.Context,
331
pc PartitionCounts,
332
request *matchingservice.QueryWorkflowRequest,
333
opts []grpc.CallOption,
334
) (*matchingservice.QueryWorkflowResponse, error) {
335
// use shallow copy since QueryRequest may contain a large payload
336
request = &matchingservice.QueryWorkflowRequest{
337
NamespaceId: request.NamespaceId,
338
TaskQueue: common.CloneProto(request.TaskQueue),
339
QueryRequest: request.QueryRequest,
340
VersionDirective: request.VersionDirective,
341
ForwardInfo: request.ForwardInfo,
342
Priority: request.Priority,
343
}
344
client, err := c.pickClientForWrite(
345
request.GetTaskQueue(),
346
request.GetNamespaceId(),
347
enumspb.TASK_QUEUE_TYPE_WORKFLOW,
348
request.GetForwardInfo().GetSourcePartition(),
349
pc,
350
)
351
if err != nil {
352
return nil, err
353
}
354
ctx, cancel := c.createContext(ctx)
355
defer cancel()
356
return client.QueryWorkflow(ctx, request, opts...)
357
}
358
359
func (c *clientImpl) DispatchNexusTask(
360
ctx context.Context,
361
request *matchingservice.DispatchNexusTaskRequest,
362
opts ...grpc.CallOption,
363
) (*matchingservice.DispatchNexusTaskResponse, error) {
364
if !isPartitionAwareKind(request.GetTaskQueue().GetKind()) {
365
return c.dispatchNexusTask(ctx, PartitionCounts{}, request, opts)
366
}
367
pkey := c.partitionCache.makeKey(
368
request.GetNamespaceId(),
369
request.GetTaskQueue().GetName(),
370
enumspb.TASK_QUEUE_TYPE_NEXUS,
371
)
372
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.dispatchNexusTask)
373
}
374
375
func (c *clientImpl) dispatchNexusTask(
376
ctx context.Context,
377
pc PartitionCounts,
378
request *matchingservice.DispatchNexusTaskRequest,
379
opts []grpc.CallOption,
380
) (*matchingservice.DispatchNexusTaskResponse, error) {
381
// use shallow copy since Request may contain a large payload
382
request = &matchingservice.DispatchNexusTaskRequest{
383
NamespaceId: request.NamespaceId,
384
TaskQueue: common.CloneProto(request.TaskQueue),
385
Request: request.Request,
386
ForwardInfo: request.ForwardInfo,
387
}
388
client, err := c.pickClientForWrite(
389
request.GetTaskQueue(),
390
request.GetNamespaceId(),
391
enumspb.TASK_QUEUE_TYPE_NEXUS,
392
request.GetForwardInfo().GetSourcePartition(),
393
pc,
394
)
395
if err != nil {
396
return nil, err
397
}
398
ctx, cancel := c.createContext(ctx)
399
defer cancel()
400
return client.DispatchNexusTask(ctx, request, opts...)
401
}
402
403
func (c *clientImpl) PollNexusTaskQueue(
404
ctx context.Context,
405
request *matchingservice.PollNexusTaskQueueRequest,
406
opts ...grpc.CallOption,
407
) (*matchingservice.PollNexusTaskQueueResponse, error) {
408
if !isPartitionAwareKind(request.GetRequest().GetTaskQueue().GetKind()) {
409
return c.pollNexusTaskQueue(ctx, PartitionCounts{}, request, opts)
410
}
411
pkey := c.partitionCache.makeKey(
412
request.GetNamespaceId(),
413
request.GetRequest().GetTaskQueue().GetName(),
414
enumspb.TASK_QUEUE_TYPE_NEXUS,
415
)
416
return invokeWithPartitionCounts(ctx, c.logger, c.partitionCache, pkey, request, opts, c.pollNexusTaskQueue)
417
}
418
419
func (c *clientImpl) pollNexusTaskQueue(
420
ctx context.Context,
421
pc PartitionCounts,
422
request *matchingservice.PollNexusTaskQueueRequest,
423
opts []grpc.CallOption,
424
) (*matchingservice.PollNexusTaskQueueResponse, error) {
425
request = common.CloneProto(request)
426
client, release, err := c.pickClientForRead(
427
request.GetRequest().GetTaskQueue(),
428
request.GetNamespaceId(),
429
enumspb.TASK_QUEUE_TYPE_NEXUS,
430
request.GetForwardedSource(),
431
pc,
432
)
433
if err != nil {
434
return nil, err
435
}
436
if release != nil {
437
defer release()
438
}
439
ctx, cancel := c.createLongPollContext(ctx)
440
defer cancel()
441
return client.PollNexusTaskQueue(ctx, request, opts...)
442
}
443
444
// processInputPartition returns a partition in certain cases that load balancer involvement is not necessary,
445
// otherwise, returns a task queue to pass down to the load balancer.
446
>
func (c *clientImpl) processInputPartition(proto *taskqueuepb.TaskQueue, nsid string, taskType enumspb.TaskQueueType, forwardedFrom string) (tqid.Partition, *tqid.TaskQueue) {
service_grpc.pb.go ×20
447
>
partition, err := tqid.PartitionFromProto(proto, nsid, taskType)
448
>
if err != nil {
449
// We preserve the old logic (not returning error in case of invalid proto info) until it's verified that
450
// clients are not sending invalid names.
451
c.logger.Info("invalid tq partition", tag.Error(err), tag.Stringer("proto", proto))
452
metrics.MatchingClientInvalidTaskQueuePartition.With(c.metricsHandler).Record(1)
453
return tqid.UnsafeTaskQueueFamily(nsid, proto.GetName()).TaskQueue(taskType).RootPartition(), nil
454
}
455
457
>
return partition, nil
458
>
}
459
461
>
case *tqid.NormalPartition:
462
>
return nil, p.TaskQueue()
463
default:
464
return partition, nil
465
}
466
}
467
468
// pickClientForWrite mutates the given proto. Callers should copy the proto before if necessary.
469
func (c *clientImpl) pickClientForWrite(
470
proto *taskqueuepb.TaskQueue,
471
nsid string,
472
taskType enumspb.TaskQueueType,
473
forwardedFrom string,
474
pc PartitionCounts,
476
>
p, tq := c.processInputPartition(proto, nsid, taskType, forwardedFrom)
477
>
if tq != nil {
478
>
p = c.loadBalancer.PickWritePartition(tq, pc)
479
>
}
480
>
proto.Name = p.RpcName()
481
>
return c.getClientForTaskQueuePartition(p)
482
}
483
484
// pickClientForRead mutates the given proto. Callers should copy the proto before if necessary.
485
func (c *clientImpl) pickClientForRead(
486
proto *taskqueuepb.TaskQueue,
487
nsid string,
488
taskType enumspb.TaskQueueType,
489
forwardedFrom string,
490
pc PartitionCounts,
491
>
) (client matchingservice.MatchingServiceClient, release func(), err error) {
service_grpc.pb.go ×20
492
>
p, tq := c.processInputPartition(proto, nsid, taskType, forwardedFrom)
493
>
if tq != nil {
494
>
token := c.loadBalancer.PickReadPartition(tq, pc)
495
>
p = token.TQPartition
496
>
release = token.Release
497
>
}
498
500
>
client, err = c.getClientForTaskQueuePartition(p)
501
>
return client, release, err
502
}
503
504
>
func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
service_grpc.pb.go ×20
505
>
return context.WithTimeout(parent, c.timeout)
506
>
}
507
508
>
func (c *clientImpl) createLongPollContext(parent context.Context) (context.Context, context.CancelFunc) {
service_grpc.pb.go ×20
509
>
return context.WithTimeout(parent, c.longPollTimeout)
510
>
}
511
513
>
spreadChange := c.spreadRouting()
514
>
spread := spreadChange.Value(p.GradualChangeKey(), time.Now())
515
>
return c.clients.Lookup(p.RoutingKey(spread))
516
>
}
517
518
func (c *clientImpl) getClientForTaskQueuePartition(
519
partition tqid.Partition,
521
>
addr, err := c.Route(partition)
522
>
if err != nil {
524
>
}
526
>
if err != nil {
527
return nil, err
528
}
530
}
531
533
>
// only normal partitions participate in scaling
534
>
return kind == enumspb.TASK_QUEUE_KIND_NORMAL
535
>
}