the pinging for fixed namespaces now works

Signed-off-by: Jakob Haahr Taankvist <[email protected]>

Author: jakobht

cmd/sharddistributor-canary/main.go

@@ -1,21 +1,27 @@
 package main
 
 import (
+	"fmt"
+	"net"
 	"os"
 	"time"
 
 	"github.com/uber-go/tally"
 	"github.com/urfave/cli/v2"
 	"go.uber.org/fx"
 	"go.uber.org/yarpc"
+	"go.uber.org/yarpc/api/peer"
 	"go.uber.org/yarpc/transport/grpc"
 	"go.uber.org/zap"
 
+	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
 	"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/canary/executors"
 	"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"
 	"github.com/uber/cadence/tools/common/commoncli"
 )
@@ -25,6 +31,7 @@ const (
 	defaultShardDistributorEndpoint = "127.0.0.1:7943"
 	defaultFixedNamespace           = "shard-distributor-canary"
 	defaultEphemeralNamespace       = "shard-distributor-canary-ephemeral"
+	defaultCanaryGRPCPort           = 7953 // Port for canary to receive ping requests
 
 	shardDistributorServiceName = "cadence-shard-distributor"
 )
@@ -33,11 +40,12 @@ func runApp(c *cli.Context) {
 	endpoint := c.String("endpoint")
 	fixedNamespace := c.String("fixed-namespace")
 	ephemeralNamespace := c.String("ephemeral-namespace")
+	canaryGRPCPort := c.Int("canary-grpc-port")
 
-	fx.New(opts(fixedNamespace, ephemeralNamespace, endpoint)).Run()
+	fx.New(opts(fixedNamespace, ephemeralNamespace, endpoint, canaryGRPCPort)).Run()
 }
 
-func opts(fixedNamespace, ephemeralNamespace, endpoint string) fx.Option {
+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},
@@ -49,22 +57,51 @@ func opts(fixedNamespace, ephemeralNamespace, endpoint string) fx.Option {
 		},
 	}
 
+	canaryGRPCAddress := fmt.Sprintf("127.0.0.1:%d", canaryGRPCPort)
+
+	// Create listener for GRPC inbound
+	listener, err := net.Listen("tcp", canaryGRPCAddress)
+	if err != nil {
+		panic(err)
+	}
+
 	transport := grpc.NewTransport()
-	yarpcConfig := yarpc.Config{
-		Name: "shard-distributor-canary",
-		Outbounds: yarpc.Outbounds{
-			shardDistributorServiceName: {
-				Unary: transport.NewSingleOutbound(endpoint),
-			},
-		},
+
+	executorMetadata := executorclient.ExecutorMetadata{
+		clientcommon.GrpcAddressMetadataKey: canaryGRPCAddress,
 	}
 
 	return fx.Options(
 		fx.Supply(
 			fx.Annotate(tally.NoopScope, fx.As(new(tally.Scope))),
 			fx.Annotate(clock.NewRealTimeSource(), fx.As(new(clock.TimeSource))),
-			yarpcConfig,
 			configuration,
+			transport,
+			executorMetadata,
+		),
+
+		fx.Provide(func(peerChooser spectatorclient.SpectatorPeerChooserInterface) 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),
+					},
+					// canary-to-canary outbound uses peer chooser to route to other canary instances
+					"shard-distributor-canary": {
+						Unary:  transport.NewOutbound(peerChooser),
+						Stream: transport.NewOutbound(peerChooser),
+					},
+				},
+			}
+		}),
+
+		fx.Provide(
+			func(t *grpc.Transport) peer.Transport { return t },
 		),
 		fx.Provide(
 			yarpc.NewDispatcher,
@@ -73,12 +110,17 @@ func opts(fixedNamespace, ephemeralNamespace, endpoint string) fx.Option {
 		fx.Provide(zap.NewDevelopment),
 		fx.Provide(log.NewLogger),
 
+		// Register canary procedures with dispatcher
+		fx.Invoke(func(dispatcher *yarpc.Dispatcher, server sharddistributorv1.ShardDistributorExecutorCanaryAPIYARPCServer) {
+			dispatcher.Register(sharddistributorv1.BuildShardDistributorExecutorCanaryAPIYARPCProcedures(server))
+		}),
+
 		// Start the YARPC dispatcher
 		fx.Invoke(func(lc fx.Lifecycle, dispatcher *yarpc.Dispatcher) {
 			lc.Append(fx.StartStopHook(dispatcher.Start, dispatcher.Stop))
 		}),
 
-		// Include the canary module
+		// 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}),
 	)
 }
@@ -110,6 +152,11 @@ func buildCLI() *cli.App {
 					Value: defaultEphemeralNamespace,
 					Usage: "namespace for ephemeral shard creation testing",
 				},
+				&cli.IntFlag{
+					Name:  "canary-grpc-port",
+					Value: defaultCanaryGRPCPort,
+					Usage: "port for canary to receive ping requests",
+				},
 			},
 			Action: func(c *cli.Context) error {
 				runApp(c)

cmd/sharddistributor-canary/main_test.go

@@ -8,5 +8,5 @@ import (
 )
 
 func TestDependenciesAreSatisfied(t *testing.T) {
-	assert.NoError(t, fx.ValidateApp(opts(defaultFixedNamespace, defaultEphemeralNamespace, defaultShardDistributorEndpoint)))
+	assert.NoError(t, fx.ValidateApp(opts(defaultFixedNamespace, defaultEphemeralNamespace, defaultShardDistributorEndpoint, defaultCanaryGRPCPort)))
 }

service/sharddistributor/canary/module.go

@@ -84,7 +84,7 @@ func opts(names NamespacesNames) fx.Option {
 		)),
 		fx.Provide(sharddistributorv1.NewFxShardDistributorExecutorCanaryAPIYARPCProcedures()),
 
-		fx.Invoke(func(lc fx.Lifecycle, chooser spectatorclient.SpectatorPeerChooserInterface, spectators spectatorclient.Spectators) {
+		fx.Invoke(func(lc fx.Lifecycle, chooser spectatorclient.SpectatorPeerChooserInterface, spectators *spectatorclient.Spectators) {
 			lc.Append(fx.Hook{
 				OnStart: func(ctx context.Context) error {
 					chooser.SetSpectators(spectators)

service/sharddistributor/canary/pinger/pingAndLog.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"time"
 
-	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
-	"github.com/uber/cadence/service/sharddistributor/client/spectatorclient"
 	"go.uber.org/yarpc"
 	"go.uber.org/zap"
+
+	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
+	"github.com/uber/cadence/service/sharddistributor/client/spectatorclient"
 )
 
 const (
@@ -26,11 +27,13 @@ func PingShard(ctx context.Context, canaryClient sharddistributorv1.ShardDistrib
 	response, err := canaryClient.Ping(ctx, request, yarpc.WithShardKey(shardKey), yarpc.WithHeader(spectatorclient.NamespaceHeader, namespace))
 	if err != nil {
 		logger.Error("Failed to ping shard", zap.String("namespace", namespace), zap.String("shard_key", shardKey), zap.Error(err))
+		return
 	}
 
 	// Verify response
 	if !response.GetOwnsShard() {
 		logger.Warn("Executor does not own shard", zap.String("namespace", namespace), zap.String("shard_key", shardKey), zap.String("executor_id", response.GetExecutorId()))
+		return
 	}
 
 	logger.Info("Successfully pinged shard owner", zap.String("namespace", namespace), zap.String("shard_key", shardKey), zap.String("executor_id", response.GetExecutorId()))

service/sharddistributor/canary/processorephemeral/canary_client_mock_test.go

@@ -13,11 +13,11 @@ import (
 
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/service/sharddistributor/client/clientcommon"
 )
 
 const (
-	NamespaceHeader        = "x-shard-distributor-namespace"
-	grpcAddressMetadataKey = "grpc_address"
+	NamespaceHeader = "x-shard-distributor-namespace"
 )
 
 // SpectatorPeerChooserInterface extends peer.Chooser with SetSpectators method
@@ -127,7 +127,7 @@ func (c *SpectatorPeerChooser) Choose(ctx context.Context, req *transport.Reques
 	}
 
 	// Extract GRPC address from owner metadata
-	grpcAddress, ok := owner.Metadata[grpcAddressMetadataKey]
+	grpcAddress, ok := owner.Metadata[clientcommon.GrpcAddressMetadataKey]
 	if !ok || grpcAddress == "" {
 		return nil, nil, yarpcerrors.InternalErrorf("no grpc_address in metadata for executor %s owning shard %s", owner.ExecutorID, req.ShardKey)
 	}

service/sharddistributor/client/spectatorclient/peer_chooser.go

@@ -12,6 +12,7 @@ import (
 	"go.uber.org/yarpc/transport/grpc"
 
 	"github.com/uber/cadence/common/log/testlogger"
+	"github.com/uber/cadence/service/sharddistributor/client/clientcommon"
 )
 
 func TestSpectatorPeerChooser_Choose_MissingShardKey(t *testing.T) {
@@ -128,7 +129,7 @@ func TestSpectatorPeerChooser_Choose_Success(t *testing.T) {
 		Return(&ShardOwner{
 			ExecutorID: "executor-1",
 			Metadata: map[string]string{
-				grpcAddressMetadataKey: "127.0.0.1:7953",
+				clientcommon.GrpcAddressMetadataKey: "127.0.0.1:7953",
 			},
 		}, nil)
 
@@ -172,7 +173,7 @@ func TestSpectatorPeerChooser_Choose_ReusesPeer(t *testing.T) {
 		Return(&ShardOwner{
 			ExecutorID: "executor-1",
 			Metadata: map[string]string{
-				grpcAddressMetadataKey: "127.0.0.1:7953",
+				clientcommon.GrpcAddressMetadataKey: "127.0.0.1:7953",
 			},
 		}, nil).Times(2)
 

service/sharddistributor/client/spectatorclient/peer_chooser_test.go

@@ -256,6 +256,7 @@ func (p *namespaceProcessor) identifyStaleExecutors(namespaceState *store.Namesp
 
 	for executorID, state := range namespaceState.Executors {
 		if now.Sub(state.LastHeartbeat) > p.cfg.HeartbeatTTL {
+			p.logger.Info("Executor has not reported a heartbeat recently", tag.ShardExecutor(executorID), tag.ShardNamespace(p.namespaceCfg.Name), tag.Value(state.LastHeartbeat))
 			expiredExecutors[executorID] = namespaceState.ShardAssignments[executorID].ModRevision
 		}
 	}