Add sharddistributor outbounds

common/config/config.go

@@ -84,6 +84,10 @@ type (
 		// To use Async APIs for a domain first specify the queue using Admin API.
 		// Either refer to one of the predefined queues in this config or alternatively specify the queue details inline in the API call.
 		AsyncWorkflowQueues map[string]AsyncWorkflowQueueProvider `yaml:"asyncWorkflowQueues"`
+		// ShardDistributorClient is the config for shard distributor client
+		// Shard distributor is used to distribute shards across multiple cadence service instances
+		// Note: This is not recommended for use, it's still experimental
+		ShardDistributorClient ShardDistributorClient `yaml:"shardDistributorClient"`
 	}
 
 	// Membership holds peer provider configuration.
@@ -592,6 +596,12 @@ type (
 		URI string `yaml:"URI"`
 	}
 
+	// ShardDistributorClient contains the config items for shard distributor
+	ShardDistributorClient struct {
+		// The host and port of the shard distributor server
+		HostPort string `yaml:"hostPort"`
+	}
+
 	// YamlNode is a lazy-unmarshaler, because *yaml.Node only exists in gopkg.in/yaml.v3, not v2,
 	// and go.uber.org/config currently uses only v2.
 	YamlNode struct {

common/constants.go

@@ -308,3 +308,8 @@ func (v FailoverType) String() string {
 		return "Unknown"
 	}
 }
+
+const (
+	ShardModeHashRing         = "hash-ring"
+	ShardModeShardDistributor = "shard-distributor"
+)

common/rpc/outbounds.go

@@ -263,3 +263,23 @@ func IsGRPCOutbound(config transport.ClientConfig) bool {
 	}
 	return namer.TransportName() == grpc.TransportName
 }
+func NewSingleGRPCOutboundBuilder(outboundName string, serviceName string, address string) OutboundsBuilder {
+	return singleGRPCOutbound{outboundName, serviceName, address}
+}
+
+type singleGRPCOutbound struct {
+	outboundName string
+	serviceName  string
+	address      string
+}
+
+func (b singleGRPCOutbound) Build(grpc *grpc.Transport, _ *tchannel.Transport) (*Outbounds, error) {
+	return &Outbounds{
+		Outbounds: yarpc.Outbounds{
+			b.outboundName: {
+				ServiceName: b.serviceName,
+				Unary:       grpc.NewSingleOutbound(b.address),
+			},
+		},
+	}, nil
+}

common/rpc/outbounds_test.go

@@ -180,6 +180,18 @@ func TestDirectOutbound(t *testing.T) {
 	assert.NotNil(t, outbounds["cadence-history"].Unary)
 }
 
+func TestSingleGRPCOutbound(t *testing.T) {
+	grpc := &grpc.Transport{}
+	tchannel := &tchannel.Transport{}
+
+	builder := NewSingleGRPCOutboundBuilder("grpc-only-out", "grpc-service-name", "http://example.com:1234")
+
+	outBound, err := builder.Build(grpc, tchannel)
+	assert.NoError(t, err)
+	assert.Equal(t, "grpc-service-name", outBound.Outbounds["grpc-only-out"].ServiceName)
+	assert.NotNil(t, outBound.Outbounds["grpc-only-out"].Unary)
+}
+
 func TestIsGRPCOutboud(t *testing.T) {
 	assert.True(t, IsGRPCOutbound(&transport.OutboundConfig{Outbounds: transport.Outbounds{Unary: (&grpc.Transport{}).NewSingleOutbound("localhost:1234")}}))
 	assert.False(t, IsGRPCOutbound(&transport.OutboundConfig{Outbounds: transport.Outbounds{Unary: (&tchannel.Transport{}).NewSingleOutbound("localhost:1234")}}))

common/rpc/params.go

@@ -133,31 +133,40 @@ func NewParams(serviceName string, config *config.Config, dc *dynamicconfig.Coll
 		}
 	}
 
-	return Params{
-		ServiceName:     serviceName,
-		HTTP:            http,
-		TChannelAddress: net.JoinHostPort(listenIP.String(), strconv.Itoa(int(serviceConfig.RPC.Port))),
-		GRPCAddress:     net.JoinHostPort(listenIP.String(), strconv.Itoa(int(serviceConfig.RPC.GRPCPort))),
-		GRPCMaxMsgSize:  serviceConfig.RPC.GRPCMaxMsgSize,
-		OutboundsBuilder: CombineOutbounds(
-			NewDirectOutboundBuilder(
-				service.History,
-				enableGRPCOutbound,
-				outboundTLS[service.History],
-				NewDirectPeerChooserFactory(service.History, logger, metricsCl),
-				dc.GetBoolProperty(dynamicconfig.EnableConnectionRetainingDirectChooser),
-			),
-			NewDirectOutboundBuilder(
-				service.Matching,
-				enableGRPCOutbound,
-				outboundTLS[service.Matching],
-				NewDirectPeerChooserFactory(service.Matching, logger, metricsCl),
-				dc.GetBoolProperty(dynamicconfig.EnableConnectionRetainingDirectChooser),
-			),
-			publicClientOutbound,
+	outboundsBuilders := []OutboundsBuilder{
+		NewDirectOutboundBuilder(
+			service.History,
+			enableGRPCOutbound,
+			outboundTLS[service.History],
+			NewDirectPeerChooserFactory(service.History, logger, metricsCl),
+			dc.GetBoolProperty(dynamicconfig.EnableConnectionRetainingDirectChooser),
+		),
+		NewDirectOutboundBuilder(
+			service.Matching,
+			enableGRPCOutbound,
+			outboundTLS[service.Matching],
+			NewDirectPeerChooserFactory(service.Matching, logger, metricsCl),
+			dc.GetBoolProperty(dynamicconfig.EnableConnectionRetainingDirectChooser),
 		),
-		InboundTLS:  inboundTLS,
-		OutboundTLS: outboundTLS,
+		publicClientOutbound,
+	}
+	if config.ShardDistributorClient.HostPort != "" {
+		outboundsBuilders = append(outboundsBuilders, NewSingleGRPCOutboundBuilder(
+			service.ShardDistributor,
+			service.ShardDistributor,
+			config.ShardDistributorClient.HostPort,
+		))
+	}
+
+	return Params{
+		ServiceName:      serviceName,
+		HTTP:             http,
+		TChannelAddress:  net.JoinHostPort(listenIP.String(), strconv.Itoa(int(serviceConfig.RPC.Port))),
+		GRPCAddress:      net.JoinHostPort(listenIP.String(), strconv.Itoa(int(serviceConfig.RPC.GRPCPort))),
+		GRPCMaxMsgSize:   serviceConfig.RPC.GRPCMaxMsgSize,
+		OutboundsBuilder: CombineOutbounds(outboundsBuilders...),
+		InboundTLS:       inboundTLS,
+		OutboundTLS:      outboundTLS,
 		InboundMiddleware: yarpc.InboundMiddleware{
 			// order matters: ForwardPartitionConfigMiddleware must be applied after ClientPartitionConfigMiddleware
 			Unary: yarpc.UnaryInboundMiddleware(&PinotComparatorMiddleware{}, &InboundMetricsMiddleware{}, &ClientPartitionConfigMiddleware{}, &ForwardPartitionConfigMiddleware{}),

common/rpc/params_test.go

@@ -38,8 +38,9 @@ func TestNewParams(t *testing.T) {
 	dc := dynamicconfig.NewNopCollection()
 	makeConfig := func(svc config.Service) *config.Config {
 		return &config.Config{
-			PublicClient: config.PublicClient{HostPort: "localhost:9999"},
-			Services:     map[string]config.Service{"frontend": svc}}
+			PublicClient:           config.PublicClient{HostPort: "localhost:9999"},
+			ShardDistributorClient: config.ShardDistributorClient{HostPort: "localhost:9998"},
+			Services:               map[string]config.Service{"frontend": svc}}
 	}
 	logger := testlogger.New(t)
 	metricsCl := metrics.NewNoopMetricsClient()

config/development.yaml

@@ -144,3 +144,6 @@ dynamicconfig:
 blobstore:
   filestore:
     outputDirectory: "/tmp/blobstore"
+
+shardDistributorClient:
+  hostPort: "localhost:7943"

host/onebox.go

@@ -35,8 +35,6 @@ import (
 	"go.uber.org/cadence/compatibility"
 	"go.uber.org/yarpc"
 	"go.uber.org/yarpc/api/transport"
-	"go.uber.org/yarpc/transport/grpc"
-	"go.uber.org/yarpc/transport/tchannel"
 
 	adminClient "github.com/uber/cadence/client/admin"
 	frontendClient "github.com/uber/cadence/client/frontend"
@@ -1138,8 +1136,8 @@ func (c *cadenceImpl) newRPCFactory(serviceName string, host membership.HostInfo
 
 		// For integration tests to generate client out of the same outbound.
 		OutboundsBuilder: rpc.CombineOutbounds(
-			&singleGRPCOutbound{testOutboundName(serviceName), serviceName, grpcAddress},
-			&singleGRPCOutbound{rpc.OutboundPublicClient, service.Frontend, frontendGrpcAddress},
+			rpc.NewSingleGRPCOutboundBuilder(testOutboundName(serviceName), serviceName, grpcAddress),
+			rpc.NewSingleGRPCOutboundBuilder(rpc.OutboundPublicClient, service.Frontend, frontendGrpcAddress),
 			rpc.NewCrossDCOutbounds(c.clusterMetadata.GetAllClusterInfo(), rpc.NewDNSPeerChooserFactory(0, c.logger)),
 			rpc.NewDirectOutboundBuilder(service.History, true, nil, directOutboundPCF, directConnRetainFn),
 			rpc.NewDirectOutboundBuilder(service.Matching, true, nil, directOutboundPCF, directConnRetainFn),
@@ -1152,23 +1150,6 @@ func testOutboundName(name string) string {
 	return "test-" + name
 }
 
-type singleGRPCOutbound struct {
-	outboundName string
-	serviceName  string
-	address      string
-}
-
-func (b singleGRPCOutbound) Build(grpc *grpc.Transport, _ *tchannel.Transport) (*rpc.Outbounds, error) {
-	return &rpc.Outbounds{
-		Outbounds: yarpc.Outbounds{
-			b.outboundName: {
-				ServiceName: b.serviceName,
-				Unary:       grpc.NewSingleOutbound(b.address),
-			},
-		},
-	}, nil
-}
-
 type versionMiddleware struct {
 }