diff --git a/chasm/interceptor_test.go b/chasm/interceptor_test.go index 346b6ccc59..4b7f235d7c 100644 --- a/chasm/interceptor_test.go +++ b/chasm/interceptor_test.go @@ -43,20 +43,35 @@ func (l *ServiceLibrary) RegisterServices(server *grpc.Server) { testspb.RegisterTestServiceServer(server, ServiceHandler{}) } -func TestChasmRequestInterceptor_ShouldRespond(t *testing.T) { +func TestChasmEngineInterceptor_ShouldRespond(t *testing.T) { ctrl := gomock.NewController(t) + mockEngine := chasm.NewMockEngine(ctrl) - mockVisibilityManager := chasm.NewMockVisibilityManager(ctrl) - requestInterceptor := chasm.ChasmRequestInterceptorProvider( + engineInterceptor := chasm.ChasmEngineInterceptorProvider( mockEngine, - mockVisibilityManager, log.NewNoopLogger(), metrics.NoopMetricsHandler, ) - server, address := startTestServer(t, grpc.UnaryInterceptor(requestInterceptor.Intercept)) + server, address := startTestServer(t, grpc.UnaryInterceptor(engineInterceptor.Intercept)) defer server.Stop() + response := testRoundTrip(t, address) + require.True(t, response.HasEngineCtx) +} + +func TestChasmVisibilityInterceptor_ShouldRespond(t *testing.T) { + ctrl := gomock.NewController(t) + + mockVisibilityManager := chasm.NewMockVisibilityManager(ctrl) + visibilityInterceptor := chasm.ChasmVisibilityInterceptorProvider(mockVisibilityManager) + + server, address := startTestServer(t, grpc.UnaryInterceptor(visibilityInterceptor.Intercept)) + defer server.Stop() + testRoundTrip(t, address) +} + +func testRoundTrip(t *testing.T, address string) *testspb.TestResponse { conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { t.Fatalf("failed to connect: %v", err) @@ -74,9 +89,9 @@ func TestChasmRequestInterceptor_ShouldRespond(t *testing.T) { RequestId: "test-request-id", }) require.NoError(t, err) - require.Equal(t, "test-request-id", response.GetRequestId()) - require.True(t, response.HasEngineCtx) + + return response } func startTestServer(t *testing.T, opt ...grpc.ServerOption) (*grpc.Server, string) { diff --git a/chasm/interceptors.go b/chasm/interceptors.go index aaa655ac06..47496f5e70 100644 --- a/chasm/interceptors.go +++ b/chasm/interceptors.go @@ -11,18 +11,17 @@ import ( const chasmRequestPrefix = "/temporal.server.chasm" -// ChasmRequestInterceptor Interceptor that intercepts RPC requests, detects Chasm-specific calls and does additional -// boilerplate processing before handing off. -type ChasmRequestInterceptor struct { +// ChasmEngineInterceptor Interceptor that intercepts RPC requests, +// detects CHASM-specific calls and does additional boilerplate processing before +// handing off. Visibility is injected separately with +// ChasmVisibilityInterceptor. +type ChasmEngineInterceptor struct { engine Engine - visibilityMgr VisibilityManager logger log.Logger metricsHandler metrics.Handler } -var _ grpc.UnaryServerInterceptor = (*ChasmRequestInterceptor)(nil).Intercept - -func (i *ChasmRequestInterceptor) Intercept( +func (i *ChasmEngineInterceptor) Intercept( ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, @@ -33,21 +32,39 @@ func (i *ChasmRequestInterceptor) Intercept( } ctx = NewEngineContext(ctx, i.engine) - ctx = NewVisibilityManagerContext(ctx, i.visibilityMgr) - return handler(ctx, req) } -func ChasmRequestInterceptorProvider( +func ChasmEngineInterceptorProvider( engine Engine, - visibilityMgr VisibilityManager, logger log.Logger, metricsHandler metrics.Handler, -) *ChasmRequestInterceptor { - return &ChasmRequestInterceptor{ +) *ChasmEngineInterceptor { + return &ChasmEngineInterceptor{ engine: engine, - visibilityMgr: visibilityMgr, logger: logger, metricsHandler: metricsHandler, } } + +// ChasmVisibilityInterceptor intercepts RPC requests and adds the CHASM +// VisibilityManager to their context. +type ChasmVisibilityInterceptor struct { + visibilityMgr VisibilityManager +} + +func (i *ChasmVisibilityInterceptor) Intercept( + ctx context.Context, + req interface{}, + info *grpc.UnaryServerInfo, + handler grpc.UnaryHandler, +) (resp interface{}, retError error) { + ctx = NewVisibilityManagerContext(ctx, i.visibilityMgr) + return handler(ctx, req) +} + +func ChasmVisibilityInterceptorProvider(visibilityMgr VisibilityManager) *ChasmVisibilityInterceptor { + return &ChasmVisibilityInterceptor{ + visibilityMgr: visibilityMgr, + } +} diff --git a/common/persistence/visibility/chasm_visibility_manager.go b/common/persistence/visibility/chasm_visibility_manager.go index c26826283c..05cbd9d2b1 100644 --- a/common/persistence/visibility/chasm_visibility_manager.go +++ b/common/persistence/visibility/chasm_visibility_manager.go @@ -28,6 +28,13 @@ func NewChasmVisibilityManager( } } +func ChasmVisibilityManagerProvider( + registry *chasm.Registry, + visibilityMgr manager.VisibilityManager, +) chasm.VisibilityManager { + return NewChasmVisibilityManager(registry, visibilityMgr) +} + // ListExecutions implements the Engine interface for visibility queries. func (e *ChasmVisibilityManager) ListExecutions( ctx context.Context, diff --git a/service/frontend/fx.go b/service/frontend/fx.go index fe9d754083..f53878ab0b 100644 --- a/service/frontend/fx.go +++ b/service/frontend/fx.go @@ -113,6 +113,8 @@ var Module = fx.Options( fx.Invoke(EndpointRegistryLifetimeHooks), fx.Provide(schedulerpb.NewSchedulerServiceLayeredClient), nexusfrontend.Module, + fx.Provide(visibility.ChasmVisibilityManagerProvider), + fx.Provide(chasm.ChasmVisibilityInterceptorProvider), ) func NewServiceProvider( @@ -213,6 +215,7 @@ func GrpcServerOptionsProvider( authInterceptor *authorization.Interceptor, maskInternalErrorDetailsInterceptor *interceptor.MaskInternalErrorDetailsInterceptor, slowRequestLoggerInterceptor *interceptor.SlowRequestLoggerInterceptor, + chasmRequestVisibilityInterceptor *chasm.ChasmVisibilityInterceptor, customInterceptors []grpc.UnaryServerInterceptor, customStreamInterceptors []grpc.StreamServerInterceptor, metricsHandler metrics.Handler, @@ -266,6 +269,7 @@ func GrpcServerOptionsProvider( sdkVersionInterceptor.Intercept, callerInfoInterceptor.Intercept, slowRequestLoggerInterceptor.Intercept, + chasmRequestVisibilityInterceptor.Intercept, } if len(customInterceptors) > 0 { // TODO: Deprecate WithChainedFrontendGrpcInterceptors and provide a inner custom interceptor @@ -736,6 +740,7 @@ func HandlerProvider( versionChecker *VersionChecker, namespaceReplicationQueue FEReplicatorNamespaceReplicationQueue, visibilityMgr manager.VisibilityManager, + chasmVisibilityMgr chasm.VisibilityManager, logger log.SnTaggedLogger, throttledLogger log.ThrottledLogger, persistenceExecutionManager persistence.ExecutionManager, diff --git a/service/history/fx.go b/service/history/fx.go index 205b6c7c44..2df294ec6f 100644 --- a/service/history/fx.go +++ b/service/history/fx.go @@ -59,12 +59,13 @@ var Module = fx.Options( fx.Provide(RateLimitInterceptorProvider), fx.Provide(HealthSignalAggregatorProvider), fx.Provide(HealthCheckInterceptorProvider), - fx.Provide(chasm.ChasmRequestInterceptorProvider), + fx.Provide(chasm.ChasmEngineInterceptorProvider), + fx.Provide(chasm.ChasmVisibilityInterceptorProvider), fx.Provide(HistoryAdditionalInterceptorsProvider), fx.Provide(service.GrpcServerOptionsProvider), fx.Provide(ESProcessorConfigProvider), fx.Provide(VisibilityManagerProvider), - fx.Provide(ChasmVisibilityManagerProvider), + fx.Provide(visibility.ChasmVisibilityManagerProvider), fx.Provide(ThrottledLoggerRpsFnProvider), fx.Provide(PersistenceRateLimitingParamsProvider), service.PersistenceLazyLoadedServiceResolverModule, @@ -208,9 +209,15 @@ func HealthCheckInterceptorProvider( } func HistoryAdditionalInterceptorsProvider( - healthCheckInterceptor *interceptor.HealthCheckInterceptor, chasmRequestInterceptor *chasm.ChasmRequestInterceptor, + healthCheckInterceptor *interceptor.HealthCheckInterceptor, + chasmRequestEngineInterceptor *chasm.ChasmEngineInterceptor, + chasmRequestVisibilityInterceptor *chasm.ChasmVisibilityInterceptor, ) []grpc.UnaryServerInterceptor { - return []grpc.UnaryServerInterceptor{healthCheckInterceptor.UnaryIntercept, chasmRequestInterceptor.Intercept} + return []grpc.UnaryServerInterceptor{ + healthCheckInterceptor.UnaryIntercept, + chasmRequestEngineInterceptor.Intercept, + chasmRequestVisibilityInterceptor.Intercept, + } } func RateLimitInterceptorProvider(