Allow for DeepHealthCheck handler to check data from interceptor

Author: laniehei

common/rpc/interceptor/health_check.go

@@ -57,7 +57,7 @@ type (
 		logger log.Logger
 	}
 
-	noopSignalAggregator struct{}
+	NoopSignalAggregator struct{}
 )
 
 // NewHealthSignalAggregatorImpl creates a new instance of HealthSignalAggregatorImpl
@@ -114,14 +114,14 @@ func isUnhealthyError(err error) bool {
 
 var NoopHealthSignalAggregator HealthSignalAggregator = newNoopSignalAggregator()
 
-func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} }
+func newNoopSignalAggregator() *NoopSignalAggregator { return &NoopSignalAggregator{} }
 
-func (a *noopSignalAggregator) Record(_ time.Duration, _ error) {}
+func (a *NoopSignalAggregator) Record(_ time.Duration, _ error) {}
 
-func (a *noopSignalAggregator) AverageLatency() float64 {
+func (a *NoopSignalAggregator) AverageLatency() float64 {
 	return 0
 }
 
-func (*noopSignalAggregator) ErrorRatio() float64 {
+func (*NoopSignalAggregator) ErrorRatio() float64 {
 	return 0
 }

service/history/fx.go

@@ -58,6 +58,7 @@ var Module = fx.Options(
 	fx.Provide(RetryableInterceptorProvider),
 	fx.Provide(TelemetryInterceptorProvider),
 	fx.Provide(RateLimitInterceptorProvider),
+	fx.Provide(HealthSignalAggregatorProvider),
 	fx.Provide(HealthCheckInterceptorProvider),
 	fx.Provide(service.GrpcServerOptionsProvider),
 	fx.Provide(ESProcessorConfigProvider),
@@ -101,6 +102,7 @@ func HandlerProvider(args NewHandlerArgs) *Handler {
 		persistenceVisibilityManager: args.PersistenceVisibilityManager,
 		persistenceHealthSignal:      args.PersistenceHealthSignal,
 		healthServer:                 args.HealthServer,
+		historyHealthSignal:          args.HistoryHealthSignal,
 		historyServiceResolver:       args.HistoryServiceResolver,
 		metricsHandler:               args.MetricsHandler,
 		payloadSerializer:            args.PayloadSerializer,
@@ -171,18 +173,26 @@ func TelemetryInterceptorProvider(
 	)
 }
 
-func HealthCheckInterceptorProvider(
+func HealthSignalAggregatorProvider(
 	dynamicCollection *dynamicconfig.Collection,
 	metricsHandler metrics.Handler,
 	logger log.ThrottledLogger,
+) *interceptor.HealthSignalAggregatorImpl {
+	return interceptor.NewHealthSignalAggregatorImpl(
+		dynamicconfig.PersistenceHealthSignalWindowSize.Get(dynamicCollection)(),
+		dynamicconfig.PersistenceHealthSignalBufferSize.Get(dynamicCollection)(),
+		logger,
+	)
+}
+
+func HealthCheckInterceptorProvider(
+	dynamicCollection *dynamicconfig.Collection,
+	healthSignalAggregator *interceptor.HealthSignalAggregatorImpl,
 ) *interceptor.HealthCheckInterceptor {
 	if dynamicconfig.HistoryHealthSignalMetricsEnabled.Get(dynamicCollection)() {
 		return interceptor.NewHealthCheckInterceptor(
-			interceptor.NewHealthSignalAggregatorImpl(
-				dynamicconfig.PersistenceHealthSignalWindowSize.Get(dynamicCollection)(),
-				dynamicconfig.PersistenceHealthSignalBufferSize.Get(dynamicCollection)(),
-				logger,
-			))
+			healthSignalAggregator,
+		)
 	}
 	return interceptor.NewHealthCheckInterceptor(
 		interceptor.NoopHealthSignalAggregator,

service/history/handler.go

@@ -37,6 +37,7 @@ import (
 	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/common/rpc/interceptor"
 	"go.temporal.io/server/common/searchattribute"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/tasktoken"
@@ -77,6 +78,7 @@ type (
 		persistenceVisibilityManager manager.VisibilityManager
 		persistenceHealthSignal      persistence.HealthSignalAggregator
 		healthServer                 *health.Server
+		historyHealthSignal          interceptor.HealthSignalAggregator
 		historyServiceResolver       membership.ServiceResolver
 		metricsHandler               metrics.Handler
 		payloadSerializer            serialization.Serializer
@@ -106,6 +108,7 @@ type (
 		PersistenceExecutionManager  persistence.ExecutionManager
 		PersistenceShardManager      persistence.ShardManager
 		PersistenceHealthSignal      persistence.HealthSignalAggregator
+		HistoryHealthSignal          interceptor.HealthSignalAggregator
 		HealthServer                 *health.Server
 		PersistenceVisibilityManager manager.VisibilityManager
 		HistoryServiceResolver       membership.ServiceResolver
@@ -206,6 +209,11 @@ func (h *Handler) DeepHealthCheck(
 		return &historyservice.DeepHealthCheckResponse{State: enumsspb.HEALTH_STATE_DECLINED_SERVING}, nil
 	}
 
+	rsp := h.checkHistoryHealthSignals()
+	if rsp != nil {
+		return rsp, nil
+	}
+
 	latency := h.persistenceHealthSignal.AverageLatency()
 	errRatio := h.persistenceHealthSignal.ErrorRatio()
 
@@ -217,6 +225,25 @@ func (h *Handler) DeepHealthCheck(
 	return &historyservice.DeepHealthCheckResponse{State: enumsspb.HEALTH_STATE_SERVING}, nil
 }
 
+// checkHistoryHealthSignal checks the history health signal that is captured by the interceptor.
+func (h *Handler) checkHistoryHealthSignals() *historyservice.DeepHealthCheckResponse {
+	// Check that the RPC latency doesn't exceed the threshold.
+	if _, ok := h.historyHealthSignal.(*interceptor.NoopSignalAggregator); ok {
+		h.logger.Warn("health signal aggregator is using noop implementation")
+	}
+	if h.historyHealthSignal.AverageLatency() > h.config.HealthRPCLatencyFailure() {
+		metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(enumsspb.HEALTH_STATE_NOT_SERVING))
+		return &historyservice.DeepHealthCheckResponse{State: enumsspb.HEALTH_STATE_NOT_SERVING}
+	}
+
+	// Check if the RPC error ratio exceeds the threshold
+	if h.historyHealthSignal.ErrorRatio() > h.config.HealthRPCErrorRatio() {
+		metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(enumsspb.HEALTH_STATE_NOT_SERVING))
+		return &historyservice.DeepHealthCheckResponse{State: enumsspb.HEALTH_STATE_NOT_SERVING}
+	}
+	return nil
+}
+
 // IsWorkflowTaskValid - whether workflow task is still valid
 func (h *Handler) IsWorkflowTaskValid(ctx context.Context, request *historyservice.IsWorkflowTaskValidRequest) (_ *historyservice.IsWorkflowTaskValidResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)