mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
Refactor deep health check code for history service and add a 60-second wait before returning unhealthy to the DeepHealthCheck (#10093)
## What changed? Pulled out the DeepHealthCheck into its own file. Added a fixed delay from startup during which DeepHealthCheck will see NOT_SERVING from the local health server as SERVING ## Why? During history service scale-up/scale-downs, individual history hosts can be detected as "NOT_SERVING" because the shard initialization hasn't happened yet. This is an expected behavior from the history service pod and does not indicate an error, but DeepHealthCheck currently treats it as one. ## How did you test it? - [x] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) ## Potential risks This change makes the DeepHealthCheck less likely to expose a problem that causes an individual history service to reprovision faster than the suppression interval. The standard HealthCheck for historyservice is unaffected. --------- Co-authored-by: Stephen Stanton <stephenstanton10@gmail.com>
This commit is contained in:
Nick Beaumont
@@ -2859,6 +2859,10 @@ Requires service restart to take effect.`,
|
||||
0.90,
|
||||
"History service health check on RPC error ratio",
|
||||
)
|
||||
HealthHistoryInitializationTime = NewGlobalDurationSetting(
|
||||
"history.healthHistoryInitializationTime",
|
||||
60*time.Second,
|
||||
"gRPC health server NOT_SERVING will be suppressed from DeepHealthCheck for this long")
|
||||
SendRawHistoryBetweenInternalServices = NewGlobalBoolSetting(
|
||||
"history.sendRawHistoryBetweenInternalServices",
|
||||
false,
|
||||
|
||||
@@ -408,6 +408,7 @@ type Config struct {
|
||||
HealthPersistenceErrorRatio dynamicconfig.FloatPropertyFn
|
||||
HealthRPCLatencyFailure dynamicconfig.FloatPropertyFn
|
||||
HealthRPCErrorRatio dynamicconfig.FloatPropertyFn
|
||||
HealthHistoryInitializationTime dynamicconfig.DurationPropertyFn
|
||||
BreakdownMetricsByTaskQueue dynamicconfig.BoolPropertyFnWithTaskQueueFilter
|
||||
|
||||
LogAllReqErrors dynamicconfig.BoolPropertyFnWithNamespaceFilter
|
||||
@@ -789,6 +790,7 @@ func NewConfig(
|
||||
HealthPersistenceErrorRatio: dynamicconfig.HealthPersistenceErrorRatio.Get(dc),
|
||||
HealthRPCLatencyFailure: dynamicconfig.HealthRPCLatencyFailure.Get(dc),
|
||||
HealthRPCErrorRatio: dynamicconfig.HealthRPCErrorRatio.Get(dc),
|
||||
HealthHistoryInitializationTime: dynamicconfig.HealthHistoryInitializationTime.Get(dc),
|
||||
|
||||
BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
|
||||
|
||||
|
||||
121
service/history/deep_health_check.go
Normal file
121
service/history/deep_health_check.go
Normal file
@@ -0,0 +1,121 @@
|
||||
package history
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
enumsspb "go.temporal.io/server/api/enums/v1"
|
||||
healthspb "go.temporal.io/server/api/health/v1"
|
||||
"go.temporal.io/server/api/historyservice/v1"
|
||||
healthcheck "go.temporal.io/server/common/health"
|
||||
"go.temporal.io/server/common/metrics"
|
||||
"go.temporal.io/server/common/persistence"
|
||||
"go.temporal.io/server/common/rpc/interceptor"
|
||||
"go.temporal.io/server/service/history/configs"
|
||||
"google.golang.org/grpc/health"
|
||||
grpchealthspb "google.golang.org/grpc/health/grpc_health_v1"
|
||||
)
|
||||
|
||||
type deepHealthCheckHandler struct {
|
||||
healthServer *health.Server
|
||||
metricsHandler metrics.Handler
|
||||
config *configs.Config
|
||||
historyHealthSignal interceptor.HealthSignalAggregator
|
||||
persistenceHealthSignal persistence.HealthSignalAggregator
|
||||
startupTime time.Time
|
||||
}
|
||||
|
||||
// DeepHealthCheck implements the grpc API from
|
||||
// ./proto/internal/temporal/server/api/historyservice/v1/request_response.proto
|
||||
func (h *deepHealthCheckHandler) DeepHealthCheck(
|
||||
ctx context.Context, now time.Time,
|
||||
) (*historyservice.DeepHealthCheckResponse, error) {
|
||||
var checks []*healthspb.HealthCheck
|
||||
|
||||
status, err := h.healthServer.Check(ctx, &grpchealthspb.HealthCheckRequest{Service: serviceName})
|
||||
if err != nil || status == nil {
|
||||
metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(enumsspb.HEALTH_STATE_NOT_SERVING))
|
||||
return &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{{
|
||||
CheckType: healthcheck.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Message: fmt.Sprintf("gRPC health check failed: %v", err),
|
||||
}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypeGRPCHealth,
|
||||
// Convert to SERVING to avoid false positives during initialization
|
||||
State: suppressStartupErrors(status.Status, now.Sub(h.startupTime), h.config.HealthHistoryInitializationTime()),
|
||||
Message: fmt.Sprintf("historyservice gRPC health check: %s", status.Status.String()),
|
||||
})
|
||||
|
||||
checks = append(checks, errorIfOverThreshold(healthcheck.CheckTypeRPCLatency,
|
||||
h.historyHealthSignal.AverageLatency(), h.config.HealthRPCLatencyFailure(),
|
||||
"historyservice latency"))
|
||||
|
||||
checks = append(checks, errorIfOverThreshold(healthcheck.CheckTypeRPCErrorRatio,
|
||||
h.historyHealthSignal.ErrorRatio(), h.config.HealthRPCErrorRatio(),
|
||||
"historyservice error ratio"))
|
||||
|
||||
checks = append(checks, errorIfOverThreshold(healthcheck.CheckTypePersistenceLatency,
|
||||
h.persistenceHealthSignal.AverageLatency(), h.config.HealthPersistenceLatencyFailure(),
|
||||
"persistenceservice latency"))
|
||||
|
||||
checks = append(checks, errorIfOverThreshold(healthcheck.CheckTypePersistenceErrRatio,
|
||||
h.persistenceHealthSignal.ErrorRatio(), h.config.HealthPersistenceErrorRatio(),
|
||||
"persistenceservice error ratio"))
|
||||
|
||||
overallState := enumsspb.HEALTH_STATE_SERVING
|
||||
|
||||
for _, check := range checks {
|
||||
if check.State == enumsspb.HEALTH_STATE_NOT_SERVING {
|
||||
overallState = check.State
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(overallState))
|
||||
|
||||
return &historyservice.DeepHealthCheckResponse{
|
||||
State: overallState,
|
||||
Checks: checks,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func suppressStartupErrors(status grpchealthspb.HealthCheckResponse_ServingStatus,
|
||||
dur time.Duration, threshold time.Duration,
|
||||
) enumsspb.HealthState {
|
||||
if dur < threshold {
|
||||
return enumsspb.HEALTH_STATE_SERVING
|
||||
}
|
||||
return toLocalHealthProto(status)
|
||||
}
|
||||
|
||||
func errorIfOverThreshold(checkType string, value float64, threshold float64, message string) *healthspb.HealthCheck {
|
||||
state := enumsspb.HEALTH_STATE_SERVING
|
||||
if value > threshold {
|
||||
state = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
}
|
||||
return &healthspb.HealthCheck{
|
||||
CheckType: checkType,
|
||||
State: state,
|
||||
Value: value,
|
||||
Threshold: threshold,
|
||||
Message: message,
|
||||
}
|
||||
}
|
||||
|
||||
func toLocalHealthProto(in grpchealthspb.HealthCheckResponse_ServingStatus) enumsspb.HealthState {
|
||||
switch in {
|
||||
case grpchealthspb.HealthCheckResponse_SERVING:
|
||||
return enumsspb.HEALTH_STATE_SERVING
|
||||
case grpchealthspb.HealthCheckResponse_NOT_SERVING:
|
||||
return enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
default:
|
||||
return enumsspb.HEALTH_STATE_UNSPECIFIED
|
||||
}
|
||||
}
|
||||
382
service/history/deep_health_check_test.go
Normal file
382
service/history/deep_health_check_test.go
Normal file
@@ -0,0 +1,382 @@
|
||||
package history
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
enumsspb "go.temporal.io/server/api/enums/v1"
|
||||
healthspb "go.temporal.io/server/api/health/v1"
|
||||
"go.temporal.io/server/api/historyservice/v1"
|
||||
health2 "go.temporal.io/server/common/health"
|
||||
"go.temporal.io/server/common/metrics"
|
||||
"go.temporal.io/server/common/persistence"
|
||||
"go.temporal.io/server/common/rpc/interceptor"
|
||||
"go.temporal.io/server/common/testing/testlogger"
|
||||
"go.temporal.io/server/service/history/configs"
|
||||
"google.golang.org/grpc/health"
|
||||
healthpb "google.golang.org/grpc/health/grpc_health_v1"
|
||||
)
|
||||
|
||||
func TestDeepHealthCheck(t *testing.T) {
|
||||
type record struct {
|
||||
latency time.Duration
|
||||
err error
|
||||
}
|
||||
|
||||
testCases := []struct {
|
||||
desc string
|
||||
timeSinceStartup time.Duration
|
||||
grpcHealthStatus healthpb.HealthCheckResponse_ServingStatus
|
||||
historyRecords []record
|
||||
persistRecords []record
|
||||
expected *historyservice.DeepHealthCheckResponse
|
||||
shouldError bool
|
||||
expectedError string
|
||||
}{
|
||||
{
|
||||
desc: "all checks healthy",
|
||||
timeSinceStartup: 5 * time.Minute,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_SERVING,
|
||||
historyRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
persistRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Message: "historyservice gRPC health check: SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "grpc not_serving suppressed within init window",
|
||||
timeSinceStartup: 30 * time.Second,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_NOT_SERVING,
|
||||
historyRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
persistRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Message: "historyservice gRPC health check: NOT_SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "rpc latency and persistence error ratio over thresholds",
|
||||
timeSinceStartup: 5 * time.Minute,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_SERVING,
|
||||
historyRecords: []record{
|
||||
{2 * time.Second, nil},
|
||||
{2 * time.Second, nil},
|
||||
},
|
||||
persistRecords: []record{
|
||||
{100 * time.Millisecond, context.DeadlineExceeded},
|
||||
{100 * time.Millisecond, context.DeadlineExceeded},
|
||||
},
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Message: "historyservice gRPC health check: SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Value: 2000,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Value: 1,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "grpc not_serving propagates after init window expires",
|
||||
timeSinceStartup: 5 * time.Minute,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_NOT_SERVING,
|
||||
historyRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
persistRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Message: "historyservice gRPC health check: NOT_SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "init window does not suppress threshold checks",
|
||||
timeSinceStartup: 30 * time.Second,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_SERVING,
|
||||
historyRecords: []record{
|
||||
{2 * time.Second, nil},
|
||||
{2 * time.Second, nil},
|
||||
},
|
||||
persistRecords: []record{
|
||||
{100 * time.Millisecond, nil},
|
||||
{100 * time.Millisecond, nil},
|
||||
},
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Message: "historyservice gRPC health check: SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Value: 2000,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 100,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "no records reports healthy (aggregator returns 0)",
|
||||
timeSinceStartup: 5 * time.Minute,
|
||||
grpcHealthStatus: healthpb.HealthCheckResponse_SERVING,
|
||||
historyRecords: nil,
|
||||
persistRecords: nil,
|
||||
expected: &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Checks: []*healthspb.HealthCheck{
|
||||
{
|
||||
CheckType: health2.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Message: "historyservice gRPC health check: SERVING",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 1000,
|
||||
Message: "historyservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypeRPCErrorRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "historyservice error ratio",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceLatency,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 1000,
|
||||
Message: "persistenceservice latency",
|
||||
},
|
||||
{
|
||||
CheckType: health2.CheckTypePersistenceErrRatio,
|
||||
State: enumsspb.HEALTH_STATE_SERVING,
|
||||
Value: 0,
|
||||
Threshold: 0.1,
|
||||
Message: "persistenceservice error ratio",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.desc, func(t *testing.T) {
|
||||
testLogger := testlogger.NewTestLogger(t, testlogger.FailOnAnyUnexpectedError)
|
||||
startupTime := time.Unix(0, 0)
|
||||
|
||||
handler := deepHealthCheckHandler{
|
||||
healthServer: health.NewServer(),
|
||||
metricsHandler: metrics.NoopMetricsHandler,
|
||||
config: &configs.Config{
|
||||
HealthPersistenceLatencyFailure: func() float64 { return 1000 },
|
||||
HealthRPCLatencyFailure: func() float64 { return 1000 },
|
||||
HealthPersistenceErrorRatio: func() float64 { return 0.1 },
|
||||
HealthRPCErrorRatio: func() float64 { return 0.1 },
|
||||
HealthHistoryInitializationTime: func() time.Duration { return time.Minute },
|
||||
},
|
||||
historyHealthSignal: interceptor.NewHealthSignalAggregator(testLogger, func() bool { return true }, time.Second, 100),
|
||||
persistenceHealthSignal: persistence.NewHealthSignalAggregator(true, time.Second, 100, metrics.NoopMetricsHandler, testLogger),
|
||||
startupTime: startupTime,
|
||||
}
|
||||
|
||||
handler.healthServer.SetServingStatus(serviceName, tc.grpcHealthStatus)
|
||||
|
||||
for _, r := range tc.historyRecords {
|
||||
handler.historyHealthSignal.Record(r.latency, r.err)
|
||||
}
|
||||
|
||||
for _, r := range tc.persistRecords {
|
||||
handler.persistenceHealthSignal.Record(1, r.latency, r.err)
|
||||
}
|
||||
|
||||
actual, err := handler.DeepHealthCheck(t.Context(), startupTime.Add(tc.timeSinceStartup))
|
||||
|
||||
if tc.shouldError && err == nil {
|
||||
require.Fail(t, "should have errored but didn't")
|
||||
}
|
||||
if err != nil {
|
||||
require.EqualError(t, err, tc.expectedError)
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.expected, actual)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -120,17 +120,22 @@ func HandlerProvider(args NewHandlerArgs) (*Handler, error) {
|
||||
}
|
||||
|
||||
handler := &Handler{
|
||||
status: common.DaemonStatusInitialized,
|
||||
config: args.Config,
|
||||
tokenSerializer: tasktoken.NewSerializer(),
|
||||
status: common.DaemonStatusInitialized,
|
||||
config: args.Config,
|
||||
tokenSerializer: tasktoken.NewSerializer(),
|
||||
deepHealthCheckHandler: deepHealthCheckHandler{
|
||||
healthServer: args.HealthServer,
|
||||
metricsHandler: args.MetricsHandler,
|
||||
config: args.Config,
|
||||
historyHealthSignal: args.HistoryHealthSignal,
|
||||
persistenceHealthSignal: args.PersistenceHealthSignal,
|
||||
startupTime: time.Now(),
|
||||
},
|
||||
logger: args.Logger,
|
||||
throttledLogger: args.ThrottledLogger,
|
||||
persistenceExecutionManager: args.PersistenceExecutionManager,
|
||||
persistenceShardManager: args.PersistenceShardManager,
|
||||
persistenceVisibilityManager: args.PersistenceVisibilityManager,
|
||||
persistenceHealthSignal: args.PersistenceHealthSignal,
|
||||
healthServer: args.HealthServer,
|
||||
historyHealthSignal: args.HistoryHealthSignal,
|
||||
historyServiceResolver: args.HistoryServiceResolver,
|
||||
metricsHandler: args.MetricsHandler,
|
||||
payloadSerializer: args.PayloadSerializer,
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"math"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/nexus-rpc/sdk-go/nexus"
|
||||
@@ -18,8 +19,6 @@ import (
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
nexuspb "go.temporal.io/api/nexus/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
enumsspb "go.temporal.io/server/api/enums/v1"
|
||||
healthspb "go.temporal.io/server/api/health/v1"
|
||||
"go.temporal.io/server/api/historyservice/v1"
|
||||
namespacespb "go.temporal.io/server/api/namespace/v1"
|
||||
persistencespb "go.temporal.io/server/api/persistence/v1"
|
||||
@@ -36,7 +35,6 @@ import (
|
||||
"go.temporal.io/server/common/convert"
|
||||
"go.temporal.io/server/common/definition"
|
||||
"go.temporal.io/server/common/headers"
|
||||
healthcheck "go.temporal.io/server/common/health"
|
||||
"go.temporal.io/server/common/log"
|
||||
"go.temporal.io/server/common/log/tag"
|
||||
"go.temporal.io/server/common/membership"
|
||||
@@ -68,7 +66,6 @@ import (
|
||||
"go.temporal.io/server/service/history/tasks"
|
||||
"go.uber.org/fx"
|
||||
"google.golang.org/grpc/health"
|
||||
grpchealthspb "google.golang.org/grpc/health/grpc_health_v1"
|
||||
)
|
||||
|
||||
type (
|
||||
@@ -82,14 +79,12 @@ type (
|
||||
tokenSerializer *tasktoken.Serializer
|
||||
config *configs.Config
|
||||
eventNotifier events.Notifier
|
||||
deepHealthCheckHandler deepHealthCheckHandler
|
||||
logger log.Logger
|
||||
throttledLogger log.Logger
|
||||
persistenceExecutionManager persistence.ExecutionManager
|
||||
persistenceShardManager persistence.ShardManager
|
||||
persistenceVisibilityManager manager.VisibilityManager
|
||||
persistenceHealthSignal persistence.HealthSignalAggregator
|
||||
healthServer *health.Server
|
||||
historyHealthSignal interceptor.HealthSignalAggregator
|
||||
historyServiceResolver membership.ServiceResolver
|
||||
metricsHandler metrics.Handler
|
||||
payloadSerializer serialization.Serializer
|
||||
@@ -208,124 +203,7 @@ func (h *Handler) DeepHealthCheck(
|
||||
ctx context.Context,
|
||||
_ *historyservice.DeepHealthCheckRequest,
|
||||
) (*historyservice.DeepHealthCheckResponse, error) {
|
||||
|
||||
var checks []*healthspb.HealthCheck
|
||||
overallState := enumsspb.HEALTH_STATE_SERVING
|
||||
|
||||
// Check 1: gRPC health (graceful shutdown / hysteresis).
|
||||
// If this fails, return early with only this check — no point running
|
||||
// metric checks if we can't even reach the gRPC health server.
|
||||
status, err := h.healthServer.Check(ctx, &grpchealthspb.HealthCheckRequest{Service: serviceName})
|
||||
if err != nil {
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypeGRPCHealth,
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Message: fmt.Sprintf("gRPC health check failed: %v", err),
|
||||
})
|
||||
metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(enumsspb.HEALTH_STATE_NOT_SERVING))
|
||||
return &historyservice.DeepHealthCheckResponse{
|
||||
State: enumsspb.HEALTH_STATE_NOT_SERVING,
|
||||
Checks: checks,
|
||||
}, nil
|
||||
}
|
||||
grpcState := enumsspb.HEALTH_STATE_SERVING
|
||||
grpcMsg := ""
|
||||
if status.Status != grpchealthspb.HealthCheckResponse_SERVING {
|
||||
grpcState = enumsspb.HEALTH_STATE_DECLINED_SERVING
|
||||
overallState = enumsspb.HEALTH_STATE_DECLINED_SERVING
|
||||
grpcMsg = "gRPC health server not serving"
|
||||
}
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypeGRPCHealth,
|
||||
State: grpcState,
|
||||
Message: grpcMsg,
|
||||
})
|
||||
|
||||
// Check 2: RPC latency
|
||||
rpcLatency := h.historyHealthSignal.AverageLatency()
|
||||
rpcLatencyThreshold := h.config.HealthRPCLatencyFailure()
|
||||
rpcLatencyState := enumsspb.HEALTH_STATE_SERVING
|
||||
rpcLatencyMsg := ""
|
||||
if rpcLatency > rpcLatencyThreshold {
|
||||
rpcLatencyState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
rpcLatencyMsg = fmt.Sprintf("RPC latency %.2fms exceeded %.2fms threshold", rpcLatency, rpcLatencyThreshold)
|
||||
if overallState != enumsspb.HEALTH_STATE_DECLINED_SERVING {
|
||||
overallState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
}
|
||||
}
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypeRPCLatency,
|
||||
State: rpcLatencyState,
|
||||
Value: rpcLatency,
|
||||
Threshold: rpcLatencyThreshold,
|
||||
Message: rpcLatencyMsg,
|
||||
})
|
||||
|
||||
// Check 3: RPC error ratio
|
||||
rpcErrRatio := h.historyHealthSignal.ErrorRatio()
|
||||
rpcErrThreshold := h.config.HealthRPCErrorRatio()
|
||||
rpcErrState := enumsspb.HEALTH_STATE_SERVING
|
||||
rpcErrMsg := ""
|
||||
if rpcErrRatio > rpcErrThreshold {
|
||||
rpcErrState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
rpcErrMsg = fmt.Sprintf("RPC error ratio %.4f exceeded %.4f threshold", rpcErrRatio, rpcErrThreshold)
|
||||
if overallState != enumsspb.HEALTH_STATE_DECLINED_SERVING {
|
||||
overallState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
}
|
||||
}
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypeRPCErrorRatio,
|
||||
State: rpcErrState,
|
||||
Value: rpcErrRatio,
|
||||
Threshold: rpcErrThreshold,
|
||||
Message: rpcErrMsg,
|
||||
})
|
||||
|
||||
// Check 4: Persistence latency
|
||||
persLatency := h.persistenceHealthSignal.AverageLatency()
|
||||
persLatencyThreshold := h.config.HealthPersistenceLatencyFailure()
|
||||
persLatencyState := enumsspb.HEALTH_STATE_SERVING
|
||||
persLatencyMsg := ""
|
||||
if persLatency > persLatencyThreshold {
|
||||
persLatencyState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
persLatencyMsg = fmt.Sprintf("Persistence latency %.2fms exceeded %.2fms threshold", persLatency, persLatencyThreshold)
|
||||
if overallState != enumsspb.HEALTH_STATE_DECLINED_SERVING {
|
||||
overallState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
}
|
||||
}
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypePersistenceLatency,
|
||||
State: persLatencyState,
|
||||
Value: persLatency,
|
||||
Threshold: persLatencyThreshold,
|
||||
Message: persLatencyMsg,
|
||||
})
|
||||
|
||||
// Check 5: Persistence error ratio
|
||||
persErrRatio := h.persistenceHealthSignal.ErrorRatio()
|
||||
persErrThreshold := h.config.HealthPersistenceErrorRatio()
|
||||
persErrState := enumsspb.HEALTH_STATE_SERVING
|
||||
persErrMsg := ""
|
||||
if persErrRatio > persErrThreshold {
|
||||
persErrState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
persErrMsg = fmt.Sprintf("Persistence error ratio %.4f exceeded %.4f threshold", persErrRatio, persErrThreshold)
|
||||
if overallState != enumsspb.HEALTH_STATE_DECLINED_SERVING {
|
||||
overallState = enumsspb.HEALTH_STATE_NOT_SERVING
|
||||
}
|
||||
}
|
||||
checks = append(checks, &healthspb.HealthCheck{
|
||||
CheckType: healthcheck.CheckTypePersistenceErrRatio,
|
||||
State: persErrState,
|
||||
Value: persErrRatio,
|
||||
Threshold: persErrThreshold,
|
||||
Message: persErrMsg,
|
||||
})
|
||||
|
||||
metrics.HistoryHostHealthGauge.With(h.metricsHandler).Record(float64(overallState))
|
||||
return &historyservice.DeepHealthCheckResponse{
|
||||
State: overallState,
|
||||
Checks: checks,
|
||||
}, nil
|
||||
return h.deepHealthCheckHandler.DeepHealthCheck(ctx, time.Now())
|
||||
}
|
||||
|
||||
// IsWorkflowTaskValid - whether workflow task is still valid
|
||||
|
||||
Reference in New Issue
Block a user