Documentation
¶
Index ¶
- Constants
- func InterceptorLogger(l zerolog.Logger) logging.Logger
- func IrrecoverableCtxInjector(signalerCtx *atomic.Pointer[irrecoverable.SignalerContext]) grpc.UnaryServerInterceptor
- func LoggingInterceptor(log zerolog.Logger) grpc.UnaryServerInterceptor
- func ShutdownStreamInterceptor(signalerCtx *atomic.Pointer[irrecoverable.SignalerContext]) grpc.StreamServerInterceptor
- func StreamLoggingInterceptor(log zerolog.Logger) grpc.StreamServerInterceptor
- type GrpcServer
- type GrpcServerBuilder
- type Option
- type RateLimiterInterceptor
Constants ¶
const DefaultGracefulStopTimeout = 5 * time.Second
DefaultGracefulStopTimeout is the time GracefulStop is allowed to wait for active streaming RPCs to finish before the server is force-stopped via Stop. Long-lived streaming subscriptions would otherwise block shutdown indefinitely.
Variables ¶
This section is empty.
Functions ¶
func InterceptorLogger ¶ added in v0.39.4
InterceptorLogger adapts a zerolog.Logger to interceptor's logging.Logger This code is simple enough to be copied and not imported.
func IrrecoverableCtxInjector ¶ added in v0.39.4
func IrrecoverableCtxInjector(signalerCtx *atomic.Pointer[irrecoverable.SignalerContext]) grpc.UnaryServerInterceptor
IrrecoverableCtxInjector injects the irrecoverable signaler context into requests so that the API handler and backend can use the irrecoverable system to handle exceptions.
It injects the signaler context into the original gRPC context via a context value using the key irrecoverable.SignalerContextKey. If signalerCtx is not set yet (because the grpc server has not finished initializing), the original context is passed unchanged.
func LoggingInterceptor ¶ added in v0.39.4
func LoggingInterceptor(log zerolog.Logger) grpc.UnaryServerInterceptor
LoggingInterceptor returns a grpc.UnaryServerInterceptor that logs incoming GRPC request and response. It extracts the requester's peer address from the gRPC context and includes it in log entries. Logs are emitted at the start and finish of each call for debugging and tracing purposes. Both start and finish logs are at debug level.
Example log messages for searching:
- Start: DBG "started call" grpc.method=Ping grpc.service=flow.access.AccessAPI peer.address=...
- Finish: DBG "finished call" grpc.method=Ping grpc.service=flow.access.AccessAPI peer.address=... grpc.code=OK grpc.time_ms=...
func ShutdownStreamInterceptor ¶ added in v0.51.1
func ShutdownStreamInterceptor(signalerCtx *atomic.Pointer[irrecoverable.SignalerContext]) grpc.StreamServerInterceptor
ShutdownStreamInterceptor cancels the per-stream context as soon as the node's irrecoverable.SignalerContext is cancelled. This lets long-lived streaming RPCs (e.g. block subscriptions) observe shutdown and return promptly, which in turn allows grpc.Server.GracefulStop to finish without waiting for the client to disconnect.
Without this interceptor, `stream.Context()` is only cancelled when the underlying transport dies; grpc.Server.GracefulStop does not cancel it. That is why active subscriptions block shutdown until either the client disconnects or the server is force-stopped via grpc.Server.Stop.
If `signalerCtx` has not been populated yet (the server is still initializing), the stream is passed through unchanged.
func StreamLoggingInterceptor ¶ added in v0.49.0
func StreamLoggingInterceptor(log zerolog.Logger) grpc.StreamServerInterceptor
StreamLoggingInterceptor returns a grpc.StreamServerInterceptor that logs incoming streaming GRPC requests. It extracts the requester's peer address from the gRPC context and includes it in log entries. Logs are emitted at the start and finish of each streaming call for debugging and tracing purposes. Both start and finish logs are at debug level.
Example log messages for searching:
- Start: DBG "started call" grpc.method=SubscribeEvents grpc.service=flow.executiondata.ExecutionDataAPI peer.address=...
- Finish: DBG "finished call" grpc.method=SubscribeEvents grpc.service=flow.executiondata.ExecutionDataAPI peer.address=... grpc.code=OK grpc.time_ms=...
Types ¶
type GrpcServer ¶
GrpcServer wraps `grpc.Server` and allows to manage it using `component.Component` interface. It can be injected into different engines making it possible to use single grpc server for multiple services which live in different modules.
func NewGrpcServer ¶
func NewGrpcServer(log zerolog.Logger, grpcListenAddr string, grpcServer *grpc.Server, grpcSignalerCtx *atomic.Pointer[irrecoverable.SignalerContext], gracefulStopTimeout time.Duration, ) *GrpcServer
NewGrpcServer returns a new grpc server. If gracefulStopTimeout is zero, DefaultGracefulStopTimeout is used.
func (*GrpcServer) GRPCAddress ¶
func (g *GrpcServer) GRPCAddress() net.Addr
GRPCAddress returns the listen address of the GRPC server. Guaranteed to be non-nil after Engine.Ready is closed.
func (*GrpcServer) RegisterService ¶ added in v0.39.4
func (g *GrpcServer) RegisterService(fn func(*grpc.Server))
RegisterService calls the provided function with the grpc server as an argument to register the server with an external service.
type GrpcServerBuilder ¶
type GrpcServerBuilder struct {
// contains filtered or unexported fields
}
GrpcServerBuilder created for separating the creation and starting GrpcServer, cause services need to be registered before the server starts.
func NewGrpcServerBuilder ¶
func NewGrpcServerBuilder( log zerolog.Logger, gRPCListenAddr string, maxRequestMsgSize uint, maxResponseMsgSize uint, rpcMetricsEnabled bool, apiRateLimits map[string]int, apiBurstLimits map[string]int, opts ...Option, ) *GrpcServerBuilder
NewGrpcServerBuilder creates a new builder for configuring and initializing a gRPC server.
The builder is configured with the provided parameters such as logger, gRPC server address, maximum message size, API rate limits, and additional options. The builder also sets up the necessary interceptors, including handling irrecoverable errors using the irrecoverable.SignalerContext. The gRPC server can be configured with options such as maximum message sizes and interceptors for handling RPC calls.
If RPC metrics are enabled, the builder adds the gRPC Prometheus interceptor for collecting metrics. Additionally, it can enable a state stream interceptor based on the configuration. Rate limiting interceptors can be added based on specified API rate limits. Logging and custom interceptors are applied, and the final gRPC server is returned.
If transport credentials are provided, a secure gRPC server is created; otherwise, an unsecured server is initialized.
Note: The gRPC server is created with the specified options and is ready for further configuration or starting.
func (*GrpcServerBuilder) Build ¶
func (b *GrpcServerBuilder) Build() *GrpcServer
type Option ¶
type Option func(*GrpcServerBuilder)
func WithGracefulStopTimeout ¶ added in v0.51.1
WithGracefulStopTimeout sets how long the server waits for active streaming RPCs to finish before force-stopping during shutdown. Defaults to DefaultGracefulStopTimeout.
func WithStreamInterceptor ¶
func WithStreamInterceptor() Option
WithStreamInterceptor sets the StreamInterceptor option to grpc server.
func WithTransportCredentials ¶
func WithTransportCredentials(transportCredentials credentials.TransportCredentials) Option
WithTransportCredentials sets the transport credentials parameters for a grpc server builder.
type RateLimiterInterceptor ¶ added in v0.39.4
type RateLimiterInterceptor struct {
// contains filtered or unexported fields
}
RateLimiterInterceptor is a gRPC interceptor that applies rate limits to incoming requests.
func NewRateLimiterInterceptor ¶ added in v0.39.4
func NewRateLimiterInterceptor(log zerolog.Logger, apiRateLimits map[string]int, apiBurstLimits map[string]int) *RateLimiterInterceptor
NewRateLimiterInterceptor creates a new rate limiter interceptor with the defined per second rate limits and the optional burst limit for each API.
func (*RateLimiterInterceptor) UnaryServerInterceptor ¶ added in v0.39.4
func (interceptor *RateLimiterInterceptor) UnaryServerInterceptor( ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (resp any, err error)
UnaryServerInterceptor returns a grpc.UnaryServerInterceptor that applies rate limits to request based on the limits defined when creating the RateLimiterInterceptor
Source Files