Vihren Atlas

activities.go ×11

Joint frontier

unlabeled · c_41f61ece5419

1 test · 2853 LOC · 139 files · introduces 1 test · 85 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
12 ranges85 lines · 2 files
Tests
1 test

Contains — complete concept membership

All code (extent)
465 ranges2853 lines · 139 files · Browse complete extent
All tests (intent)
1 testBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

1 test introduced at this concept.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

2 files ranked by introduced lines: 85 introduced LOC across 12 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/migration/activities.go 79 introduced LOC · 11 ranges

Open complete file

257
258 // checkReplicationOnce checks whether the remote cluster has caught up on all shards.
259 > func (a *activities) checkReplicationOnce(ctx context.Context, waitRequest WaitReplicationRequest) (bool, error) { activities.go
260 > resp, err := a.HistoryClient.GetReplicationStatus(ctx, &historyservice.GetReplicationStatusRequest{
261 > RemoteClusters: []string{waitRequest.RemoteCluster}, // only the specified remote cluster
262 > })
263 > if err != nil {
264 return false, err
265 }
266
267 > localShards := resp.Shards activities.go
268 >
269 > if int(waitRequest.ShardCount) != len(localShards) {
270 return false, fmt.Errorf("GetReplicationStatus returns %d shards, expecting %d", len(resp.Shards), waitRequest.ShardCount)
271 }
272
273 > sort.SliceStable(localShards, func(i, j int) bool { activities.go
274 return localShards[i].ShardId < localShards[j].ShardId
275 })
276
277 // this is the minimum task ID each shard must reach before catchup is considered complete
278 > requiredMinTaskIDPerShard := waitRequest.WaitForTaskIds activities.go
279 >
280 > shardStatuses := make([]shardStatus, 0, len(localShards))
281 >
282 > for _, localShard := range localShards {
283 > remoteShardProgress, hasRemoteShardProgress := localShard.RemoteClusters[waitRequest.RemoteCluster]
284 > if !hasRemoteShardProgress {
285 a.Logger.Info("GetReplicationStatus response missing expected remote cluster for shard during replication catchup",
286 tag.ShardID(localShard.ShardId),
293 }
294
295 > shardStatuses = append(shardStatuses, classifyShardReplicationStatus( activities.go
296 > localShard,
297 > remoteShardProgress,
298 > requiredMinTaskIDPerShard[localShard.ShardId],
299 > waitRequest.AllowedLaggingTasks,
300 > waitRequest.AllowedLagging,
301 > ))
302 }
303
304 > var ( activities.go
305 > readyShardCount int
306 > notReadyShardCount int
307 > noWatermarkShardCount int
308 >
309 > maxLaggingTasks int64
310 > maxTimeLag time.Duration
311 > )
312 > // -1 (not a valid shard id) means no shard had genuine lag, e.g. when the not-ready
313 > // shards are all no-watermark and thus carry zero lag.
314 > maxLaggingTasksShardID := int32(-1)
315 > maxTimeLagShardID := int32(-1)
316 >
317 > for _, status := range shardStatuses {
318 > if status.isReady {
319 readyShardCount++
320 continue
321 }
322
323 > notReadyShardCount++ activities.go
324 >
325 > // No-ack-watermark shards carry zero lag values
326 > if status.laggingTasks == 0 {
327 > noWatermarkShardCount++
328 > }
329
330 > if status.laggingTasks > maxLaggingTasks { activities.go
331 maxLaggingTasks = status.laggingTasks
332 maxLaggingTasksShardID = status.shardID
333 }
334
335 > if status.timeLag > maxTimeLag { activities.go
336 maxTimeLag = status.timeLag
337 maxTimeLagShardID = status.shardID
340
341 // emit metrics about how many shards are ready
342 > a.MetricsHandler.Gauge(metrics.CatchUpReadyShardCountGauge.Name()).Record( activities.go
343 > float64(readyShardCount),
344 > metrics.OperationTag(metrics.MigrationWorkflowScope),
345 > metrics.NamespaceTag(waitRequest.Namespace),
346 > metrics.TargetClusterTag(waitRequest.RemoteCluster))
347 >
348 > // emit the not-ready shard count (namespace-tagged) so the catchup failure mode is
349 > // observable per namespace during a failover.
350 > a.MetricsHandler.Gauge(metrics.CatchUpNotReadyShardCountGauge.Name()).Record(
351 > float64(notReadyShardCount),
352 > metrics.OperationTag(metrics.MigrationWorkflowScope),
353 > metrics.NamespaceTag(waitRequest.Namespace),
354 > metrics.TargetClusterTag(waitRequest.RemoteCluster))
355 >
356 > isReady := notReadyShardCount == 0
357 >
358 > if !isReady {
359 > a.Logger.Info("Wait catchup not ready",
360 > tag.String("RemoteCluster", waitRequest.RemoteCluster),
361 > tag.String("Namespace", waitRequest.Namespace),
362 > tag.Int("TotalShards", len(localShards)),
363 > tag.Int("ReadyShards", readyShardCount),
364 > tag.Int("NotReadyShards", len(localShards)-readyShardCount),
365 > tag.Int("NoWatermarkShards", noWatermarkShardCount),
366 > tag.Duration("AllowedLagging", waitRequest.AllowedLagging),
367 > tag.Int64("AllowedLaggingTasks", waitRequest.AllowedLaggingTasks),
368 > tag.Int32("MaxLaggingTasksShardID", maxLaggingTasksShardID),
369 > tag.Int64("MaxLaggingTasks", maxLaggingTasks),
370 > tag.Int32("MaxTimeLagShardID", maxTimeLagShardID),
371 > tag.Duration("MaxTimeLag", maxTimeLag),
372 > )
373 > }
374
375 > return isReady, nil activities.go
376 }
377
go.temporal.io/server/common/metrics/metrics_mock.go 6 introduced LOC · 1 range

Open complete file

57
58 // Gauge mocks base method.
59 > func (m *MockHandler) Gauge(arg0 string) GaugeIface { metrics_mock.go
60 > m.ctrl.T.Helper()
61 > ret := m.ctrl.Call(m, "Gauge", arg0)
62 > ret0, _ := ret[0].(GaugeIface)
63 > return ret0
64 > }
65
66 // Gauge indicates an expected call of Gauge.

Direct hierarchy

Children