feat(shard-distributor): add shard handover latency metrics

[arzonus]

What changed?
Added metrics to track shard assignment distribution and handover latency in the shard distributor service. This includes:

• New ShardDistributorShardAssignmentDistributionLatency metric measuring time from shard assignment to distribution
• New ShardDistributorShardHandoverLatency metric measuring handover time between executors
• HandoverType enum (GRACEFUL/EMERGENCY) to distinguish handover types

Why?
To provide visibility into shard distribution performance and identify potential issues with shard handovers, particularly distinguishing between graceful and emergency handovers.

How did you test it?
Added comprehensive unit tests covering metric emission logic and shard statistics tracking.

Potential risks
Additional storage operations during heartbeat processing (mitigated by running metrics emission in background goroutine).

Release notes
Added shard handover latency metrics for monitoring distribution performance.

Documentation Changes
None required.

[jakobht]

Suggested change

	if len(newAssignedShardIDs) == 0 {
	// no handovers happened, nothing to do
	return
	}

Nit: This is not necessary, the loop below will just not run if the list is empty

[jakobht]

We actually already have this function in common/convert.go

[jakobht]

This is not done in the background I think. Did you mean to spawn a go-routine?

[jakobht]

Reading from nil returns empty, false so we don't need this check

Suggested change

	// check if handover stats exist at all
	isShardHandoverStatsExists := assignedState.ShardHandoverStats != nil
	for _, shardID := range newAssignedShardIDs {
	if !isShardHandoverStatsExists {
	// no handover stats at all, means no handovers happened before
	continue
	}
	for _, shardID := range newAssignedShardIDs {

[jakobht]

I found this function hard to follow. Maybe we can do something like this:

Suggested change

	func filterNewlyAssignedShardIDs(previousHeartbeat *store.HeartbeatState, assignedState *store.AssignedState) []string {
	// if assignedState is nil, no shards are assigned
	if assignedState == nil || len(assignedState.AssignedShards) == 0 {
	return nil
	}
	// if previousHeartbeat is nil, all assigned shards are new
	if previousHeartbeat == nil {
	var newAssignedShardIDs = make([]string, len(assignedState.AssignedShards))
	var i int
	for assignedShardID := range assignedState.AssignedShards {
	newAssignedShardIDs[i] = assignedShardID
	i++
	}
	return newAssignedShardIDs
	}
	// find shards that are assigned now but were not reported in the previous heartbeat
	var newAssignedShardIDs []string
	for assignedShardID := range assignedState.AssignedShards {
	if _, ok := previousHeartbeat.ReportedShards[assignedShardID]; !ok {
	newAssignedShardIDs = append(newAssignedShardIDs, assignedShardID)
	}
	}
	return newAssignedShardIDs
	}
	func filterNewlyAssignedShardIDs(previousHeartbeat *store.HeartbeatState, assignedState *store.AssignedState) []string {
	if assignedState == nil || len(assignedState.AssignedShards) == 0 {
	return nil
	}
	newAssignedShardIDs := make([]string, 0)
	for assignedShardID := range assignedState.AssignedShards {
	if previousHeartbeat == nil || !shardInReportedShards(previousHeartbeat.ReportedShards, assignedShardID) {
	newAssignedShardIDs = append(newAssignedShardIDs, assignedShardID)
	}
	}
	return newAssignedShardIDs
	}
	func shardInReportedShards(reportedShards map[string]*types.ShardStatusReport, shardID string) bool {
	_, ok := reportedShards[shardID]
	return ok
	}

[jakobht]

nit: i don't think we need this comment

[jakobht]

Nit: the other tests in the file does a slightly different table test setup, lets keep it consistent.

[jakobht]

We do not use ctrl anywhere

[jakobht]

Lets use tally.NewTestScope instead of the old testify mocks. E.g. like here: https://github.com/cadence-workflow/cadence/blob/master/service/sharddistributor/wrappers/metered/metered_test.go#L80

[jakobht]

I think we should consider having some helpers to generate the AssignedShards and ShardHandoverStats. e.g:

func makeAssignedShards(shardIDs ...string) map[string]*types.ShardAssignment {
    m := make(map[string]*types.ShardAssignment, len(shardIDs))
    for _, id := range shardIDs {
        // Default to READY since that is the common case in your tests
        m[id] = &types.ShardAssignment{Status: types.AssignmentStatusREADY}
    }
    return m
}

[jakobht]

The previousHeartbeat is the same in every test. I think we should have a fixed value in the test code - we already tested the filter function in the other test