feat: [Executor-client] Passthrough logic for migration (#7345)

<!-- Describe what has changed in this PR -->
**What changed?**
- Implemented the logic for the migration:
    - no heartbeating if local passthrough
    - no assignment if local passthrough shadow
    - assignment in case of distributed passthrough and onboarded
- logic for consistency checks and metrics in case passthrough shadow
and distributed passthrough
- static configuration to control the first step of the migration


<!-- Tell your future self why have you made these changes -->
**Why?**
These mechanism are needed to migrate incrementally from using sharding
logic which is part of the system to the shard distributor


<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
unit tests and local testing

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
nothing is onboarded yet, except canary service

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

---------

Signed-off-by: edigregorio <[email protected]>

Author: eleonoradgr

cmd/sharddistributor-canary/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/service/sharddistributor/canary"
+	"github.com/uber/cadence/service/sharddistributor/config"
 	"github.com/uber/cadence/service/sharddistributor/executorclient"
 	"github.com/uber/cadence/tools/common/commoncli"
 )
@@ -38,8 +39,8 @@ func runApp(c *cli.Context) {
 func opts(fixedNamespace, ephemeralNamespace, endpoint string) fx.Option {
 	config := executorclient.Config{
 		Namespaces: []executorclient.NamespaceConfig{
-			{Namespace: fixedNamespace, HeartBeatInterval: 1 * time.Second},
-			{Namespace: ephemeralNamespace, HeartBeatInterval: 1 * time.Second},
+			{Namespace: fixedNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeONBOARDED},
+			{Namespace: ephemeralNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeONBOARDED},
 		},
 	}
 

common/types/sharddistributor.go

@@ -24,7 +24,7 @@ package types
 
 import "fmt"
 
-//go:generate enumer -type=ExecutorStatus,ShardStatus,AssignmentStatus -json -output sharddistributor_statuses_enumer_generated.go
+//go:generate enumer -type=ExecutorStatus,ShardStatus,AssignmentStatus,MigrationMode -json -output sharddistributor_statuses_enumer_generated.go
 
 type GetShardOwnerRequest struct {
 	ShardKey  string
@@ -204,9 +204,9 @@ const (
 type MigrationMode int32
 
 const (
-	MigrationModeINVALID                = 0
-	MigrationModeLOCALPASSTHROUGH       = 1
-	MigrationModeLOCALPASSTHROUGHSHADOW = 2
-	MigrationModeDISTRIBUTEDPASSTHROUGH = 3
-	MigrationModeONBOARDED              = 4
+	MigrationModeINVALID                MigrationMode = 0
+	MigrationModeLOCALPASSTHROUGH       MigrationMode = 1
+	MigrationModeLOCALPASSTHROUGHSHADOW MigrationMode = 2
+	MigrationModeDISTRIBUTEDPASSTHROUGH MigrationMode = 3
+	MigrationModeONBOARDED              MigrationMode = 4
 )

common/types/sharddistributor_statuses_enumer_generated.go

@@ -95,7 +95,8 @@ const (
 	MigrationModeONBOARDED              = "onboarded"
 )
 
-var configMode = map[string]types.MigrationMode{
+// ConfigMode maps string migration mode values to types.MigrationMode
+var ConfigMode = map[string]types.MigrationMode{
 	MigrationModeINVALID:                types.MigrationModeINVALID,
 	MigrationModeLOCALPASSTHROUGH:       types.MigrationModeLOCALPASSTHROUGH,
 	MigrationModeLOCALPASSTHROUGHSHADOW: types.MigrationModeLOCALPASSTHROUGHSHADOW,
@@ -106,11 +107,11 @@ var configMode = map[string]types.MigrationMode{
 func (s *ShardDistribution) GetMigrationMode(namespace string) types.MigrationMode {
 	for _, ns := range s.Namespaces {
 		if ns.Name == namespace {
-			return configMode[ns.Mode]
+			return ConfigMode[ns.Mode]
 		}
 	}
 	// TODO in the dynamic configuration I will setup a default value
-	return configMode[MigrationModeONBOARDED]
+	return ConfigMode[MigrationModeONBOARDED]
 }
 
 // NewConfig returns new service config with default values

service/sharddistributor/config/config.go

@@ -112,6 +112,7 @@ func newExecutorWithConfig[SP ShardProcessor](params Params[SP], namespaceConfig
 		timeSource:             params.TimeSource,
 		stopC:                  make(chan struct{}),
 		metrics:                metricsScope,
+		migrationMode:          namespaceConfig.GetMigrationMode(),
 	}, nil
 }
 

service/sharddistributor/executorclient/client.go

@@ -67,6 +67,7 @@ type executorImpl[SP ShardProcessor] struct {
 	processLoopWG          sync.WaitGroup
 	assignmentMutex        sync.Mutex
 	metrics                tally.Scope
+	migrationMode          types.MigrationMode
 }
 
 func (e *executorImpl[SP]) Start(ctx context.Context) {
@@ -102,6 +103,12 @@ func (e *executorImpl[SP]) AssignShardsFromLocalLogic(ctx context.Context, shard
 }
 
 func (e *executorImpl[SP]) heartbeatloop(ctx context.Context) {
+	// Check if initial migration mode is LOCAL_PASSTHROUGH - if so, skip heartbeating entirely
+	if e.migrationMode == types.MigrationModeLOCALPASSTHROUGH {
+		e.logger.Info("initial migration mode is local passthrough, skipping heartbeat loop")
+		return
+	}
+
 	heartBeatTimer := e.timeSource.NewTimer(getJitteredHeartbeatDuration(e.heartBeatInterval, heartbeatJitterMax))
 	defer heartBeatTimer.Stop()
 
@@ -117,11 +124,39 @@ func (e *executorImpl[SP]) heartbeatloop(ctx context.Context) {
 			return
 		case <-heartBeatTimer.Chan():
 			heartBeatTimer.Reset(getJitteredHeartbeatDuration(e.heartBeatInterval, heartbeatJitterMax))
-			shardAssignment, err := e.heartbeat(ctx)
+			shardAssignment, migrationMode, err := e.heartbeat(ctx)
 			if err != nil {
 				e.logger.Error("failed to heartbeat", tag.Error(err))
 				continue // TODO: should we stop the executor, and drop all the shards?
 			}
+
+			// Handle migration mode logic
+			switch migrationMode {
+			case types.MigrationModeLOCALPASSTHROUGH:
+				// LOCAL_PASSTHROUGH: statically assigned, stop heartbeating
+				e.logger.Info("local passthrough mode: stopping heartbeat loop")
+				return
+
+			case types.MigrationModeLOCALPASSTHROUGHSHADOW:
+				// LOCAL_PASSTHROUGH_SHADOW: check response but don't apply it
+				e.compareAssignments(shardAssignment)
+				continue
+
+			case types.MigrationModeDISTRIBUTEDPASSTHROUGH:
+				// DISTRIBUTED_PASSTHROUGH: validate then apply the assignment
+				e.compareAssignments(shardAssignment)
+				// Continue with applying the assignment from heartbeat
+
+			case types.MigrationModeONBOARDED:
+				// ONBOARDED: normal flow, apply the assignment from heartbeat
+				// Continue with normal assignment logic below
+
+			default:
+				e.logger.Warn("unknown migration mode, skipping assignment",
+					tag.Dynamic("migration-mode", migrationMode))
+				continue
+			}
+
 			if !e.assignmentMutex.TryLock() {
 				e.logger.Warn("already doing shard assignment, will skip this assignment")
 				e.metrics.Counter(metricsconstants.ShardDistributorExecutorAssignmentSkipped).Inc(1)
@@ -141,7 +176,7 @@ func (e *executorImpl[SP]) heartbeatloop(ctx context.Context) {
 	}
 }
 
-func (e *executorImpl[SP]) heartbeat(ctx context.Context) (shardAssignments map[string]*types.ShardAssignment, err error) {
+func (e *executorImpl[SP]) heartbeat(ctx context.Context) (shardAssignments map[string]*types.ShardAssignment, migrationMode types.MigrationMode, err error) {
 	// Fill in the shard status reports
 	shardStatusReports := make(map[string]*types.ShardStatusReport)
 	e.managedProcessors.Range(func(shardID string, managedProcessor *managedProcessor[SP]) bool {
@@ -169,10 +204,19 @@ func (e *executorImpl[SP]) heartbeat(ctx context.Context) (shardAssignments map[
 	// Send the request
 	response, err := e.shardDistributorClient.Heartbeat(ctx, request)
 	if err != nil {
-		return nil, fmt.Errorf("send heartbeat: %w", err)
+		return nil, types.MigrationModeINVALID, fmt.Errorf("send heartbeat: %w", err)
+	}
+
+	previousMode := e.migrationMode
+	currentMode := response.MigrationMode
+	if previousMode != currentMode {
+		e.logger.Info("migration mode transition",
+			tag.Dynamic("previous", previousMode),
+			tag.Dynamic("current", currentMode))
+		e.migrationMode = currentMode
 	}
 
-	return response.ShardAssignments, nil
+	return response.ShardAssignments, response.MigrationMode, nil
 }
 
 func (e *executorImpl[SP]) updateShardAssignment(ctx context.Context, shardAssignments map[string]*types.ShardAssignment) {
@@ -246,6 +290,52 @@ func (e *executorImpl[SP]) stopShardProcessors() {
 	wg.Wait()
 }
 
+// compareAssignments compares the local assignments with the heartbeat response assignments
+// and emits convergence or divergence metrics
+func (e *executorImpl[SP]) compareAssignments(heartbeatAssignments map[string]*types.ShardAssignment) {
+	// Get current local assignments
+	localAssignments := make(map[string]bool)
+	e.managedProcessors.Range(func(shardID string, managedProcessor *managedProcessor[SP]) bool {
+		if managedProcessor.getState() == processorStateStarted {
+			localAssignments[shardID] = true
+		}
+		return true
+	})
+
+	// Check if all local assignments are in heartbeat assignments with READY status
+	for shardID := range localAssignments {
+		assignment, exists := heartbeatAssignments[shardID]
+		if !exists || assignment.Status != types.AssignmentStatusREADY {
+			e.logger.Warn("assignment divergence: local shard not in heartbeat or not ready",
+				tag.Dynamic("shard-id", shardID))
+			e.emitMetricsConvergence(false)
+			return
+		}
+	}
+
+	// Check if all heartbeat READY assignments are in local assignments
+	for shardID, assignment := range heartbeatAssignments {
+		if assignment.Status == types.AssignmentStatusREADY {
+			if !localAssignments[shardID] {
+				e.logger.Warn("assignment divergence: heartbeat shard not in local",
+					tag.Dynamic("shard-id", shardID))
+				e.emitMetricsConvergence(false)
+				return
+			}
+		}
+	}
+
+	e.emitMetricsConvergence(true)
+}
+
+func (e *executorImpl[SP]) emitMetricsConvergence(converged bool) {
+	if converged {
+		e.metrics.Counter(metricsconstants.ShardDistributorExecutorAssignmentConvergence).Inc(1)
+	} else {
+		e.metrics.Counter(metricsconstants.ShardDistributorExecutorAssignmentDivergence).Inc(1)
+	}
+}
+
 func getJitteredHeartbeatDuration(interval time.Duration, jitterMax time.Duration) time.Duration {
 	jitterMaxNanos := int64(jitterMax)
 	randomJitterNanos := rand.Int63n(jitterMaxNanos)