Wire up canary ping/pong components in module

Integrate all canary components into the module:
- Create SpectatorPeerChooser and manage its lifecycle
- Provide canary client using the dispatcher
- Create and start the pinger component
- Register ping handler as a YARPC server
- Wire spectators to peer chooser on startup

This connects all the pieces needed for executor-to-executor
canary testing via ping/pong requests.

Dependencies:
- Requires SpectatorPeerChooser
- Requires PingHandler
- Requires Pinger component

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

Author: jakobht

service/sharddistributor/canary/module.go

@@ -1,15 +1,21 @@
 package canary
 
 import (
+	"context"
+
 	"go.uber.org/fx"
+	"go.uber.org/yarpc"
 
 	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
 	"github.com/uber/cadence/service/sharddistributor/canary/executors"
 	"github.com/uber/cadence/service/sharddistributor/canary/factory"
+	"github.com/uber/cadence/service/sharddistributor/canary/handler"
+	"github.com/uber/cadence/service/sharddistributor/canary/pinger"
 	"github.com/uber/cadence/service/sharddistributor/canary/processor"
 	"github.com/uber/cadence/service/sharddistributor/canary/processorephemeral"
 	"github.com/uber/cadence/service/sharddistributor/canary/sharddistributorclient"
 	"github.com/uber/cadence/service/sharddistributor/client/executorclient"
+	"github.com/uber/cadence/service/sharddistributor/client/spectatorclient"
 )
 
 type NamespacesNames struct {
@@ -49,5 +55,42 @@ func opts(names NamespacesNames) fx.Option {
 		executors.Module(names.FixedNamespace, names.EphemeralNamespace, names.ExternalAssignmentNamespace),
 
 		processorephemeral.ShardCreatorModule([]string{names.EphemeralNamespace}),
+
+		spectatorclient.Module(),
+		fx.Provide(spectatorclient.NewSpectatorPeerChooser),
+		fx.Invoke(func(chooser spectatorclient.SpectatorPeerChooserInterface, lc fx.Lifecycle) {
+			lc.Append(fx.StartStopHook(chooser.Start, chooser.Stop))
+		}),
+
+		// Create canary client using the dispatcher's client config
+		fx.Provide(func(dispatcher *yarpc.Dispatcher) sharddistributorv1.ShardDistributorExecutorCanaryAPIYARPCClient {
+			config := dispatcher.ClientConfig("shard-distributor-canary")
+			return sharddistributorv1.NewShardDistributorExecutorCanaryAPIYARPCClient(config)
+		}),
+
+		fx.Provide(func(params pinger.Params) *pinger.Pinger {
+			return pinger.NewPinger(params, names.FixedNamespace, 32)
+		}),
+		fx.Invoke(func(p *pinger.Pinger, lc fx.Lifecycle) {
+			lc.Append(fx.StartStopHook(p.Start, p.Stop))
+		}),
+
+		// Register canary ping handler to receive ping requests from other executors
+		fx.Provide(handler.NewPingHandler),
+		fx.Provide(fx.Annotate(
+			func(h *handler.PingHandler) sharddistributorv1.ShardDistributorExecutorCanaryAPIYARPCServer {
+				return h
+			},
+		)),
+		fx.Provide(sharddistributorv1.NewFxShardDistributorExecutorCanaryAPIYARPCProcedures()),
+
+		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)
+					return nil
+				},
+			})
+		}),
 	)
 }

service/sharddistributor/canary/module_test.go

@@ -23,7 +23,8 @@ func TestModule(t *testing.T) {
 	// Create mocks
 	ctrl := gomock.NewController(t)
 	mockClientConfig := transporttest.NewMockClientConfig(ctrl)
-	outbound := grpc.NewTransport().NewOutbound(yarpctest.NewFakePeerList())
+	transport := grpc.NewTransport()
+	outbound := transport.NewOutbound(yarpctest.NewFakePeerList())
 
 	mockClientConfig.EXPECT().Caller().Return("test-executor").Times(2)
 	mockClientConfig.EXPECT().Service().Return("shard-distributor").Times(2)
@@ -43,6 +44,11 @@ func TestModule(t *testing.T) {
 		},
 	}
 
+	// Create a mock dispatcher
+	dispatcher := yarpc.NewDispatcher(yarpc.Config{
+		Name: "test-canary",
+	})
+
 	// Create a test app with the library, check that it starts and stops
 	fxtest.New(t,
 		fx.Supply(
@@ -52,6 +58,8 @@ func TestModule(t *testing.T) {
 			fx.Annotate(mockClientConfigProvider, fx.As(new(yarpc.ClientConfig))),
 			zaptest.NewLogger(t),
 			config,
+			transport,
+			dispatcher,
 		),
 		Module(NamespacesNames{FixedNamespace: "shard-distributor-canary", EphemeralNamespace: "shard-distributor-canary-ephemeral", ExternalAssignmentNamespace: "test-external-assignment", SharddistributorServiceName: "cadence-shard-distributor"}),
 	).RequireStart().RequireStop()