feat(shard-distributor): load-based rebalancing and placement

cmd/sharddistributor-canary/main.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/uber-go/tally"
+	"github.com/uber-go/tally/prometheus"
 	"github.com/urfave/cli/v2"
 	"go.uber.org/fx"
 	"go.uber.org/yarpc"
@@ -16,13 +17,15 @@ import (
 
 	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
 	"github.com/uber/cadence/common/clock"
+	cadenceconfig "github.com/uber/cadence/common/config"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/service/sharddistributor/canary"
 	"github.com/uber/cadence/service/sharddistributor/canary/executors"
+	"github.com/uber/cadence/service/sharddistributor/canary/replay"
 	"github.com/uber/cadence/service/sharddistributor/client/clientcommon"
 	"github.com/uber/cadence/service/sharddistributor/client/executorclient"
 	"github.com/uber/cadence/service/sharddistributor/client/spectatorclient"
-	"github.com/uber/cadence/service/sharddistributor/config"
+	sdconfig "github.com/uber/cadence/service/sharddistributor/config"
 	"github.com/uber/cadence/tools/common/commoncli"
 )
 
@@ -32,6 +35,8 @@ const (
 	defaultFixedNamespace           = "shard-distributor-canary"
 	defaultEphemeralNamespace       = "shard-distributor-canary-ephemeral"
 	defaultCanaryGRPCPort           = 7953 // Port for canary to receive ping requests
+	defaultCanaryMetricsPort        = 9098
+	defaultReplayNamespace          = "shard-distributor-replay"
 
 	shardDistributorServiceName = "cadence-shard-distributor"
 )
@@ -41,21 +46,52 @@ func runApp(c *cli.Context) {
 	fixedNamespace := c.String("fixed-namespace")
 	ephemeralNamespace := c.String("ephemeral-namespace")
 	canaryGRPCPort := c.Int("canary-grpc-port")
+	canaryMetricsPort := c.Int("canary-metrics-port")
 
-	fx.New(opts(fixedNamespace, ephemeralNamespace, endpoint, canaryGRPCPort)).Run()
+	replayOpts := replay.Options{
+		CSVPath:           c.String("replay-csv"),
+		Speed:             c.Float64("replay-speed"),
+		Namespace:         c.String("replay-namespace"),
+		NumFixedExecutors: c.Int("replay-num-fixed-executors"),
+	}
+
+	fx.New(opts(fixedNamespace, ephemeralNamespace, endpoint, canaryGRPCPort, canaryMetricsPort, replayOpts)).Run()
 }
 
-func opts(fixedNamespace, ephemeralNamespace, endpoint string, canaryGRPCPort int) fx.Option {
-	configuration := clientcommon.Config{
-		Namespaces: []clientcommon.NamespaceConfig{
-			{Namespace: fixedNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeONBOARDED},
-			{Namespace: ephemeralNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeONBOARDED},
-			{Namespace: executors.LocalPassthroughNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeLOCALPASSTHROUGH},
-			{Namespace: executors.LocalPassthroughShadowNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeLOCALPASSTHROUGHSHADOW},
-			{Namespace: executors.DistributedPassthroughNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeDISTRIBUTEDPASSTHROUGH},
-			{Namespace: executors.ExternalAssignmentNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: config.MigrationModeDISTRIBUTEDPASSTHROUGH},
+func opts(fixedNamespace, ephemeralNamespace, endpoint string, canaryGRPCPort int, canaryMetricsPort int, replayOpts replay.Options) fx.Option {
+	logger, _ := zap.NewDevelopment()
+	cadenceLogger := log.NewLogger(logger)
+
+	metricsConfig := cadenceconfig.Metrics{
+		Prometheus: &prometheus.Configuration{
+			ListenAddress: fmt.Sprintf("127.0.0.1:%d", canaryMetricsPort),
+			TimerType:     "histogram",
 		},
 	}
+	metricsScope := metricsConfig.NewScope(cadenceLogger, "shard-distributor-canary")
+
+	if replayOpts.Namespace == "" {
+		replayOpts.Namespace = defaultReplayNamespace
+	}
+	if replayOpts.NumFixedExecutors <= 0 {
+		replayOpts.NumFixedExecutors = 3
+	}
+
+	configuration := clientcommon.Config{}
+	if replayOpts.Enabled() {
+		configuration.Namespaces = []clientcommon.NamespaceConfig{
+			{Namespace: replayOpts.Namespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeONBOARDED},
+		}
+	} else {
+		configuration.Namespaces = []clientcommon.NamespaceConfig{
+			{Namespace: fixedNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeONBOARDED},
+			{Namespace: ephemeralNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeONBOARDED},
+			{Namespace: executors.LocalPassthroughNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeLOCALPASSTHROUGH},
+			{Namespace: executors.LocalPassthroughShadowNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeLOCALPASSTHROUGHSHADOW},
+			{Namespace: executors.DistributedPassthroughNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeDISTRIBUTEDPASSTHROUGH},
+			{Namespace: executors.ExternalAssignmentNamespace, HeartBeatInterval: 1 * time.Second, MigrationMode: sdconfig.MigrationModeDISTRIBUTEDPASSTHROUGH},
+		}
+	}
 
 	canaryGRPCAddress := fmt.Sprintf("127.0.0.1:%d", canaryGRPCPort)
 
@@ -71,16 +107,35 @@ func opts(fixedNamespace, ephemeralNamespace, endpoint string, canaryGRPCPort in
 		clientcommon.GrpcAddressMetadataKey: canaryGRPCAddress,
 	}
 
-	return fx.Options(
+	options := []fx.Option{
 		fx.Supply(
-			fx.Annotate(tally.NoopScope, fx.As(new(tally.Scope))),
+			fx.Annotate(metricsScope, fx.As(new(tally.Scope))),
 			fx.Annotate(clock.NewRealTimeSource(), fx.As(new(clock.TimeSource))),
 			configuration,
 			transport,
 			executorMetadata,
+			logger,
+			replayOpts,
 		),
+	}
 
-		fx.Provide(func(peerChooser spectatorclient.SpectatorPeerChooserInterface) yarpc.Config {
+	if replayOpts.Enabled() {
+		options = append(options, fx.Provide(func() yarpc.Config {
+			return yarpc.Config{
+				Name: "shard-distributor-canary",
+				Inbounds: yarpc.Inbounds{
+					transport.NewInbound(listener), // Listen for incoming ping requests
+				},
+				Outbounds: yarpc.Outbounds{
+					shardDistributorServiceName: {
+						Unary:  transport.NewSingleOutbound(endpoint),
+						Stream: transport.NewSingleOutbound(endpoint),
+					},
+				},
+			}
+		}))
+	} else {
+		options = append(options, fx.Provide(func(peerChooser spectatorclient.SpectatorPeerChooserInterface) yarpc.Config {
 			return yarpc.Config{
 				Name: "shard-distributor-canary",
 				Inbounds: yarpc.Inbounds{
@@ -98,17 +153,18 @@ func opts(fixedNamespace, ephemeralNamespace, endpoint string, canaryGRPCPort in
 					},
 				},
 			}
-		}),
+		}))
+	}
 
+	options = append(options,
 		fx.Provide(
 			func(t *grpc.Transport) peer.Transport { return t },
 		),
 		fx.Provide(
 			yarpc.NewDispatcher,
 			func(d *yarpc.Dispatcher) yarpc.ClientConfig { return d }, // Reprovide the dispatcher as a client config
+			func(l *zap.Logger) log.Logger { return log.NewLogger(l) },
 		),
-		fx.Provide(zap.NewDevelopment),
-		fx.Provide(log.NewLogger),
 
 		// We do decorate instead of Invoke because we want to start and stop the dispatcher at the
 		// correct time.
@@ -130,8 +186,18 @@ func opts(fixedNamespace, ephemeralNamespace, endpoint string, canaryGRPCPort in
 		}),
 
 		// Include the canary module - it will set up spectator peer choosers and canary client
-		canary.Module(canary.NamespacesNames{FixedNamespace: fixedNamespace, EphemeralNamespace: ephemeralNamespace, ExternalAssignmentNamespace: executors.ExternalAssignmentNamespace, SharddistributorServiceName: shardDistributorServiceName}),
+		canary.ModuleWithReplay(
+			canary.NamespacesNames{
+				FixedNamespace:              fixedNamespace,
+				EphemeralNamespace:          ephemeralNamespace,
+				ExternalAssignmentNamespace: executors.ExternalAssignmentNamespace,
+				SharddistributorServiceName: shardDistributorServiceName,
+			},
+			replayOpts,
+		),
 	)
+
+	return fx.Options(options...)
 }
 
 func buildCLI() *cli.App {
@@ -166,6 +232,30 @@ func buildCLI() *cli.App {
 					Value: defaultCanaryGRPCPort,
 					Usage: "port for canary to receive ping requests",
 				},
+				&cli.IntFlag{
+					Name:  "canary-metrics-port",
+					Value: defaultCanaryMetricsPort,
+					Usage: "port for canary Prometheus metrics",
+				},
+				&cli.StringFlag{
+					Name:  "replay-csv",
+					Usage: "enable CSV load replay (path to CSV: timestamp,load0,...,loadN-1)",
+				},
+				&cli.StringFlag{
+					Name:  "replay-namespace",
+					Value: defaultReplayNamespace,
+					Usage: "fixed namespace used for CSV replay",
+				},
+				&cli.IntFlag{
+					Name:  "replay-num-fixed-executors",
+					Value: 3,
+					Usage: "number of fixed-namespace executors to run in-process during replay",
+				},
+				&cli.Float64Flag{
+					Name:  "replay-speed",
+					Value: 1.0,
+					Usage: "CSV replay speed multiplier (timestamp-following mode)",
+				},
 			},
 			Action: func(c *cli.Context) error {
 				runApp(c)

common/metrics/defs.go

@@ -2984,6 +2984,24 @@ const (
 	ShardDistributorAssignLoopFail
 
 	ShardDistributorActiveShards
+	ShardDistributorActiveExecutors
+	// ShardDistributorAssignmentLoadMaxOverMean measures max/mean across executor loads
+	ShardDistributorAssignmentLoadMaxOverMean
+	// ShardDistributorAssignmentLoadCV measures coefficient of variation across executor loads
+	ShardDistributorAssignmentLoadCV
+	// ShardDistributorAssignmentSmoothedLoadMaxOverMean measures max/mean across executor smoothed loads
+	ShardDistributorAssignmentSmoothedLoadMaxOverMean
+	// ShardDistributorAssignmentSmoothedLoadCV measures coefficient of variation across executor smoothed loads
+	ShardDistributorAssignmentSmoothedLoadCV
+	// ShardDistributorAssignmentReportedLoadMissingRatio measures the fraction of assigned shards that have no
+	// per-shard load report on the currently assigned executor.
+	ShardDistributorAssignmentReportedLoadMissingRatio
+	// ShardDistributorAssignmentSmoothedLoadMissingRatio measures the fraction of assigned shards that are missing
+	// smoothed load statistics (ShardStats entry absent or no update time).
+	ShardDistributorAssignmentSmoothedLoadMissingRatio
+	// ShardDistributorAssignmentSmoothedLoadStaleRatio measures the fraction of assigned shards whose smoothed load
+	// statistics are stale relative to the leader's staleness threshold.
+	ShardDistributorAssignmentSmoothedLoadStaleRatio
 	ShardDistributorTotalExecutors
 	ShardDistributorOldestExecutorHeartbeatLag
 
@@ -2992,6 +3010,19 @@ const (
 	ShardDistributorStoreRequestsPerNamespace
 	ShardDistributorStoreLatencyHistogramPerNamespace
 
+	// ShardDistributorLoadBalanceCycles reports number of load-balance cycles executed (counter).
+	ShardDistributorLoadBalanceCycles
+	// ShardDistributorLoadBalanceMoves reports number of shards moved by the load balancer (counter).
+	ShardDistributorLoadBalanceMoves
+	// ShardDistributorLoadBalanceSourceExecutorsInitial reports the count of source executors (above upper band)
+	// at the start of the load-balance cycle.
+	ShardDistributorLoadBalanceSourceExecutorsInitial
+	// ShardDistributorLoadBalanceDestinationExecutorsInitial reports the count of destination executors (below lower band)
+	// at the start of the load-balance cycle.
+	ShardDistributorLoadBalanceDestinationExecutorsInitial
+	// ShardDistributorLoadBalanceStopReason increments once per cycle with a low-cardinality `reason` tag describing why load-balance stopped.
+	ShardDistributorLoadBalanceStopReason
+
 	// ShardDistributorShardAssignmentDistributionLatency measures the time taken between assignment of a shard
 	// and the time it is fully distributed to executors
 	ShardDistributorShardAssignmentDistributionLatency
@@ -3789,7 +3820,25 @@ var MetricDefs = map[ServiceIdx]map[MetricIdx]metricDefinition{
 		ShardDistributorAssignLoopSuccess:               {metricName: "shard_distrubutor_shard_assign_success", metricType: Counter},
 		ShardDistributorAssignLoopFail:                  {metricName: "shard_distrubutor_shard_assign_fail", metricType: Counter},
 
-		ShardDistributorActiveShards:               {metricName: "shard_distributor_active_shards", metricType: Gauge},
+		ShardDistributorActiveShards:                      {metricName: "shard_distributor_active_shards", metricType: Gauge},
+		ShardDistributorActiveExecutors:                   {metricName: "shard_distributor_active_executors", metricType: Gauge},
+		ShardDistributorAssignmentLoadMaxOverMean:         {metricName: "shard_distributor_assignment_load_max_over_mean", metricType: Gauge},
+		ShardDistributorAssignmentLoadCV:                  {metricName: "shard_distributor_assignment_load_cv", metricType: Gauge},
+		ShardDistributorAssignmentSmoothedLoadMaxOverMean: {metricName: "shard_distributor_assignment_smoothed_load_max_over_mean", metricType: Gauge},
+		ShardDistributorAssignmentSmoothedLoadCV:          {metricName: "shard_distributor_assignment_smoothed_load_cv", metricType: Gauge},
+		ShardDistributorAssignmentReportedLoadMissingRatio: {
+			metricName: "shard_distributor_assignment_reported_load_missing_ratio",
+			metricType: Gauge,
+		},
+		ShardDistributorAssignmentSmoothedLoadMissingRatio: {
+			metricName: "shard_distributor_assignment_smoothed_load_missing_ratio",
+			metricType: Gauge,
+		},
+		ShardDistributorAssignmentSmoothedLoadStaleRatio: {
+			metricName: "shard_distributor_assignment_smoothed_load_stale_ratio",
+			metricType: Gauge,
+		},
+
 		ShardDistributorTotalExecutors:             {metricName: "shard_distributor_total_executors", metricType: Gauge},
 		ShardDistributorOldestExecutorHeartbeatLag: {metricName: "shard_distributor_oldest_executor_heartbeat_lag", metricType: Gauge},
 
@@ -3798,6 +3847,23 @@ var MetricDefs = map[ServiceIdx]map[MetricIdx]metricDefinition{
 		ShardDistributorStoreRequestsPerNamespace:         {metricName: "shard_distributor_store_requests_per_namespace", metricType: Counter},
 		ShardDistributorStoreLatencyHistogramPerNamespace: {metricName: "shard_distributor_store_latency_histogram_per_namespace", metricType: Histogram, buckets: ShardDistributorExecutorStoreLatencyBuckets},
 
+		ShardDistributorLoadBalanceCycles: {metricName: "shard_distributor_load_balance_cycles", metricType: Counter},
+		ShardDistributorLoadBalanceMoves:  {metricName: "shard_distributor_load_balance_moves", metricType: Counter},
+		ShardDistributorLoadBalanceSourceExecutorsInitial: {
+			metricName: "shard_distributor_load_balance_source_executors_initial",
+			metricType: Gauge,
+		},
+		ShardDistributorLoadBalanceDestinationExecutorsInitial: {
+			metricName: "shard_distributor_load_balance_destination_executors_initial",
+			metricType: Gauge,
+		},
+		ShardDistributorLoadBalanceStopReason: {
+			metricName: "shard_distributor_load_balance_stop_reason",
+			metricType: Counter,
+		},
+
+		ShardDistributorShardAssignmentDistributionLatency: {metricName: "shard_distributor_shard_assignment_distribution_latency", metricType: Histogram, buckets: Default1ms100s.buckets()},
+		ShardDistributorShardHandoverLatency:               {metricName: "shard_distributor_shard_handover_latency", metricType: Histogram, buckets: Default1ms100s.buckets()},
 		ShardDistributorShardAssignmentDistributionLatency: {metricName: "shard_distributor_shard_assignment_distribution_latency", metricType: Histogram, buckets: ShardDistributorShardAssignmentLatencyBuckets},
 		ShardDistributorShardHandoverLatency:               {metricName: "shard_distributor_shard_handover_latency", metricType: Histogram, buckets: ShardDistributorShardAssignmentLatencyBuckets},
 	},

config/development.yaml

@@ -158,6 +158,11 @@ shardDistribution:
       type: fixed
       mode: onboarded
       shardNum: 32
+    - name: shard-distributor-replay
+      type: fixed
+      mode: onboarded
+      # Must match the CSV column count when using canary replay mode.
+      shardNum: 1000
     - name: shard-distributor-canary-ephemeral
       mode: onboarded
       type: ephemeral
@@ -186,6 +191,12 @@ shardDistribution:
   process:
     period: 1s
     heartbeatTTL: 2s
+    loadBalance:
+      disableBenefitGating: false
+      severeImbalanceRatio: 1.3
+      moveBudgetProportion: 0.05
+      hysteresisUpperBand: 1.1
+      hysteresisLowerBand: 0.95
 
 shard-distributor-matching:
   namespaces:
@@ -194,3 +205,4 @@ shard-distributor-matching:
       migration_mode: local_pass
       ttl_shard: 5m
       ttl_report: 1m
+

config/dynamicconfig/development.yaml

@@ -93,3 +93,7 @@ shardDistributor.migrationMode:
   - value: "distributed_pass"
     constraints:
       namespace: "test-external-assignment"
+shardDistributor.loadBalancingMode:
+  - value: "greedy"
+    constraints:
+      namespace: "shard-distributor-replay"

docker/grafana/provisioning/datasources/default.yaml

@@ -1,6 +1,6 @@
 datasources:
   - name: Prometheus
     type: prometheus
-    url: http://host.docker.internal:9090
+    url: http://prometheus:9090
     version: 1
     editable: true

docker/prometheus/prometheus.yml

@@ -1,5 +1,5 @@
 global:
-  scrape_interval: 5s
+  scrape_interval: 1s
   external_labels:
     monitor: 'cadence-monitor'
 scrape_configs:
@@ -11,4 +11,5 @@ scrape_configs:
           - 'cadence:8001'
           - 'cadence:8002'
           - 'cadence:8003'
-          - 'host.docker.internal:8004' # Endpoint for Cadence samples running on localhost 
+          - 'host.docker.internal:8004' # Cadence shard distributor metrics exposed on host
+          - 'host.docker.internal:9098' # Shard distributor canary metrics exposed on host

service/sharddistributor/canary/executors/executors.go

@@ -107,3 +107,21 @@ func Module(fixedNamespace, ephemeralNamespace, externalAssignmentNamespace stri
 		fx.Invoke(NewExecutorsModule),
 	)
 }
+
+// ReplayFixedModule starts N fixed-namespace executors for a single namespace.
+// It is used by the CSV replay canary to simulate multiple executors in a single process.
+func ReplayFixedModule(namespace string, fixedExecutorCount int) fx.Option {
+	if fixedExecutorCount <= 0 {
+		fixedExecutorCount = 1
+	}
+
+	var options []fx.Option
+	for i := 0; i < fixedExecutorCount; i++ {
+		options = append(options, fx.Provide(func(params executorclient.Params[*processor.ShardProcessor]) (ExecutorResult, error) {
+			return NewExecutorWithFixedNamespace(params, namespace)
+		}))
+	}
+	options = append(options, fx.Invoke(NewExecutorsModule))
+
+	return fx.Module("ReplayFixedExecutors", fx.Options(options...))
+}