mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Lanie HeiandGitHub
ba116f7db7
This adds an interceptor to the handler service that logs latency and errors from handlers, and can be reused in other services. Adds the interceptor to the history service, that takes a health aggregator provider. This provider is also added to the handler such that we can read these values in the DeepHealthCheck handler within the history service. A future PR will add this check to the other services.
48 lines
1.2 KiB
Go
48 lines
1.2 KiB
Go
package interceptor
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync/atomic"
|
|
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/server/common/api"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
// HealthInterceptor rejects frontend requests if the host has not been marked as healthy.
|
|
type (
|
|
HealthInterceptor struct {
|
|
healthy atomic.Bool
|
|
}
|
|
)
|
|
|
|
var _ grpc.UnaryServerInterceptor = (*HealthInterceptor)(nil).Intercept
|
|
|
|
var notHealthyErr = serviceerror.NewUnavailable("Frontend is not healthy yet")
|
|
|
|
// NewHealthInterceptor returns a new HealthInterceptor. It starts with state not healthy.
|
|
func NewHealthInterceptor() *HealthInterceptor {
|
|
return &HealthInterceptor{}
|
|
}
|
|
|
|
func (i *HealthInterceptor) Intercept(
|
|
ctx context.Context,
|
|
req interface{},
|
|
info *grpc.UnaryServerInfo,
|
|
handler grpc.UnaryHandler,
|
|
) (interface{}, error) {
|
|
// only enforce health check on WorkflowService and OperatorService
|
|
if strings.HasPrefix(info.FullMethod, api.WorkflowServicePrefix) ||
|
|
strings.HasPrefix(info.FullMethod, api.OperatorServicePrefix) {
|
|
if !i.healthy.Load() {
|
|
return nil, notHealthyErr
|
|
}
|
|
}
|
|
return handler(ctx, req)
|
|
}
|
|
|
|
func (i *HealthInterceptor) SetHealthy(healthy bool) {
|
|
i.healthy.Store(healthy)
|
|
}
|