Revert change to history handler, add interceptor

Author: laniehei

common/rpc/interceptor/health_check.go

@@ -0,0 +1,127 @@
+package interceptor
+
+import (
+	"context"
+	"time"
+
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/aggregate"
+	"go.temporal.io/server/common/log"
+	"google.golang.org/grpc"
+)
+
+// HealthCheckInterceptor is a gRPC interceptor that records health metrics
+type HealthCheckInterceptor struct {
+	historyHealthSignalAggregator HealthSignalAggregator
+}
+
+// NewHealthCheckInterceptor creates a new health check interceptor
+func NewHealthCheckInterceptor(historyHealthSignalAggregator HealthSignalAggregator) *HealthCheckInterceptor {
+	return &HealthCheckInterceptor{
+		historyHealthSignalAggregator: historyHealthSignalAggregator,
+	}
+}
+
+// UnaryIntercept implements the gRPC unary interceptor interface
+func (h *HealthCheckInterceptor) UnaryIntercept(
+	ctx context.Context,
+	req interface{},
+	info *grpc.UnaryServerInfo,
+	handler grpc.UnaryHandler,
+) (interface{}, error) {
+	startTime := time.Now()
+	resp, err := handler(ctx, req)
+	elapsed := time.Since(startTime)
+
+	h.historyHealthSignalAggregator.Record(elapsed, err)
+
+	return resp, err
+}
+
+type (
+	// HealthSignalAggregator interface for tracking RPC health signals
+	HealthSignalAggregator interface {
+		Record(latency time.Duration, err error)
+		AverageLatency() float64
+		ErrorRatio() float64
+	}
+
+	// HealthSignalAggregatorImpl implements HealthSignalAggregator
+	HealthSignalAggregatorImpl struct {
+		status int32
+
+		latencyAverage aggregate.MovingWindowAverage
+		errorRatio     aggregate.MovingWindowAverage
+
+		logger log.Logger
+	}
+
+	noopSignalAggregator struct{}
+)
+
+// NewHealthSignalAggregatorImpl creates a new instance of HealthSignalAggregatorImpl
+func NewHealthSignalAggregatorImpl(
+	windowSize time.Duration,
+	maxBufferSize int,
+	logger log.Logger,
+) *HealthSignalAggregatorImpl {
+	ret := &HealthSignalAggregatorImpl{
+		latencyAverage: aggregate.NewMovingWindowAvgImpl(windowSize, maxBufferSize),
+		errorRatio:     aggregate.NewMovingWindowAvgImpl(windowSize, maxBufferSize),
+		logger:         logger,
+	}
+	return ret
+}
+
+func (s *HealthSignalAggregatorImpl) Record(latency time.Duration, err error) {
+	s.latencyAverage.Record(latency.Milliseconds())
+
+	if isUnhealthyError(err) {
+		s.errorRatio.Record(1)
+	} else {
+		s.errorRatio.Record(0)
+	}
+}
+
+func (s *HealthSignalAggregatorImpl) AverageLatency() float64 {
+	return s.latencyAverage.Average()
+}
+
+func (s *HealthSignalAggregatorImpl) ErrorRatio() float64 {
+	return s.errorRatio.Average()
+}
+
+func isUnhealthyError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if common.IsContextCanceledErr(err) {
+		return true
+	}
+	if common.IsContextDeadlineExceededErr(err) {
+		return true
+	}
+
+	switch err.(type) {
+	case *serviceerror.Unavailable,
+		*serviceerror.DeadlineExceeded,
+		*serviceerror.Canceled:
+		return true
+	}
+	return false
+}
+
+var NoopHealthSignalAggregator HealthSignalAggregator = newNoopSignalAggregator()
+
+func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} }
+
+func (a *noopSignalAggregator) Record(_ time.Duration, _ error) {}
+
+func (a *noopSignalAggregator) AverageLatency() float64 {
+	return 0
+}
+
+func (*noopSignalAggregator) ErrorRatio() float64 {
+	return 0
+}

service/history/fx.go

@@ -58,12 +58,12 @@ var Module = fx.Options(
 	fx.Provide(RetryableInterceptorProvider),
 	fx.Provide(TelemetryInterceptorProvider),
 	fx.Provide(RateLimitInterceptorProvider),
+	fx.Provide(HealthCheckInterceptorProvider),
 	fx.Provide(service.GrpcServerOptionsProvider),
 	fx.Provide(ESProcessorConfigProvider),
 	fx.Provide(VisibilityManagerProvider),
 	fx.Provide(ThrottledLoggerRpsFnProvider),
 	fx.Provide(PersistenceRateLimitingParamsProvider),
-	fx.Provide(HealthSignalAggregatorProvider),
 	service.PersistenceLazyLoadedServiceResolverModule,
 	fx.Provide(ServiceResolverProvider),
 	fx.Provide(EventNotifierProvider),
@@ -116,7 +116,6 @@ func HandlerProvider(args NewHandlerArgs) *Handler {
 		taskQueueManager:             args.TaskQueueManager,
 		taskCategoryRegistry:         args.TaskCategoryRegistry,
 		dlqMetricsEmitter:            args.DLQMetricsEmitter,
-		healthSignalAggregator:       args.HealthSignalAggregator,
 
 		replicationTaskFetcherFactory:    args.ReplicationTaskFetcherFactory,
 		replicationTaskConverterProvider: args.ReplicationTaskConverterFactory,
@@ -172,6 +171,23 @@ func TelemetryInterceptorProvider(
 	)
 }
 
+func HealthCheckInterceptorProvider(
+	dynamicCollection *dynamicconfig.Collection,
+	metricsHandler metrics.Handler,
+	logger log.ThrottledLogger,
+) *interceptor.HealthCheckInterceptor {
+	if dynamicconfig.HistoryHealthSignalMetricsEnabled.Get(dynamicCollection)() {
+		return interceptor.NewHealthCheckInterceptor(
+			interceptor.NewHealthSignalAggregatorImpl(
+				dynamicconfig.PersistenceHealthSignalWindowSize.Get(dynamicCollection)(),
+				dynamicconfig.PersistenceHealthSignalBufferSize.Get(dynamicCollection)(),
+				logger,
+			))
+	}
+	return interceptor.NewHealthCheckInterceptor(
+		interceptor.NoopHealthSignalAggregator,
+	)
+}
 func RateLimitInterceptorProvider(
 	serviceConfig *configs.Config,
 ) *interceptor.RateLimitInterceptor {
@@ -292,22 +308,3 @@ func ReplicationProgressCacheProvider(
 ) replication.ProgressCache {
 	return replication.NewProgressCache(serviceConfig, logger, handler)
 }
-
-func HealthSignalAggregatorProvider(
-	dynamicCollection *dynamicconfig.Collection,
-	metricsHandler metrics.Handler,
-	logger log.ThrottledLogger,
-) HealthSignalAggregator {
-	if dynamicconfig.HistoryHealthSignalMetricsEnabled.Get(dynamicCollection)() {
-		return NewHealthSignalAggregatorImpl(
-			dynamicconfig.PersistenceHealthSignalAggregationEnabled.Get(dynamicCollection)(),
-			dynamicconfig.PersistenceHealthSignalWindowSize.Get(dynamicCollection)(),
-			dynamicconfig.PersistenceHealthSignalBufferSize.Get(dynamicCollection)(),
-			metricsHandler,
-			dynamicconfig.ShardRPSWarnLimit.Get(dynamicCollection),
-			dynamicconfig.ShardPerNsRPSWarnPercent.Get(dynamicCollection),
-			logger,
-		)
-	}
-	return NoopHealthSignalAggregator
-}

service/history/handler.go

@@ -106,7 +106,7 @@ type (
 		ThrottledLogger              log.ThrottledLogger
 		PersistenceExecutionManager  persistence.ExecutionManager
 		PersistenceShardManager      persistence.ShardManager
-		PersistenceHealthSignal      persistence.historyHealthSignalAggregator
+		PersistenceHealthSignal      persistence.HealthSignalAggregator
 		HealthServer                 *health.Server
 		PersistenceVisibilityManager manager.VisibilityManager
 		HistoryServiceResolver       membership.ServiceResolver
@@ -206,20 +206,6 @@ func (h *Handler) DeepHealthCheck(
 	if status.Status != healthpb.HealthCheckResponse_SERVING {
 		return &historyservice.DeepHealthCheckResponse{State: enumsspb.HEALTH_STATE_DECLINED_SERVING}, nil
 	}
-	// Check that the RPC latency doesn't exceed the threshold.
-	if _, ok := h.historyHealthSignalAggregator.(*noopSignalAggregator); ok {
-		h.logger.Warn("health signal aggregator is using noop implementation")
-	}
-	if h.historyHealthSignalAggregator.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}, nil
-	}
-
-	// Check if the RPC error ratio exceeds the threshold
-	if h.historyHealthSignalAggregator.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}, nil
-	}
 
 	// Check if the persistence layer is healthy.
 	latency := h.persistenceHealthSignal.AverageLatency()