cosmos-sdk/baseapp/grpcserver.go
Aaron Craelius f2d4a98039
feat: OpenTelemetry configuration and BaseApp instrumentation (#25516)
Co-authored-by: Tyler <48813565+technicallyty@users.noreply.github.com>
Co-authored-by: Alex | Cosmos Labs <alex@cosmoslabs.io>
2025-12-10 23:15:18 +00:00

129 lines
4.8 KiB
Go

package baseapp
import (
"context"
"fmt"
"strconv"
gogogrpc "github.com/cosmos/gogoproto/grpc"
grpcmiddleware "github.com/grpc-ecosystem/go-grpc-middleware"
grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
errorsmod "cosmossdk.io/errors"
storetypes "cosmossdk.io/store/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
grpctypes "github.com/cosmos/cosmos-sdk/types/grpc"
)
// RegisterGRPCServer registers gRPC services directly with the gRPC server.
func (app *BaseApp) RegisterGRPCServer(server gogogrpc.Server) {
app.RegisterGRPCServerWithSkipCheckHeader(server, false)
}
// RegisterGRPCServerWithSkipCheckHeader registers gRPC services with the specified gRPC server
// and bypass check header flag. During the commit phase, gRPC queries may be processed before the block header
// is fully updated, causing header checks to fail erroneously. Skipping the header check in these cases prevents
// false negatives and ensures more robust query handling. While bypassing the header check is generally preferred to avoid false
// negatives during the commit phase, there are niche scenarios where someone might want to enable it.
// For instance, if an application requires strict validation to ensure that the query context exactly
// reflects the expected block header (for consistency or security reasons), then enabling header checks
// could be beneficial. However, this strictness comes at the cost of potentially more frequent errors
// when queries occur during the commit phase.
func (app *BaseApp) RegisterGRPCServerWithSkipCheckHeader(server gogogrpc.Server, skipCheckHeader bool) {
// Define an interceptor for all gRPC queries: this interceptor will create
// a new sdk.Context, and pass it into the query handler.
interceptor := func(grpcCtx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
// If there's some metadata in the context, retrieve it.
md, ok := metadata.FromIncomingContext(grpcCtx)
if !ok {
return nil, status.Error(codes.Internal, "unable to retrieve metadata")
}
// Get height header from the request context, if present.
var height int64
if heightHeaders := md.Get(grpctypes.GRPCBlockHeightHeader); len(heightHeaders) == 1 {
height, err = strconv.ParseInt(heightHeaders[0], 10, 64)
if err != nil {
return nil, errorsmod.Wrapf(
sdkerrors.ErrInvalidRequest,
"Baseapp.RegisterGRPCServer: invalid height header %q: %v", grpctypes.GRPCBlockHeightHeader, err)
}
if err := checkNegativeHeight(height); err != nil {
return nil, err
}
}
// Create the sdk.Context. Passing false as 2nd arg, as we can't
// actually support proofs with gRPC right now.
sdkCtx, err := app.CreateQueryContextWithCheckHeader(height, false, !skipCheckHeader)
if err != nil {
return nil, err
}
// Add relevant gRPC headers
if height == 0 {
height = sdkCtx.BlockHeight() // If height was not set in the request, set it to the latest
}
md = metadata.Pairs(grpctypes.GRPCBlockHeightHeader, strconv.FormatInt(height, 10))
if err = grpc.SetHeader(grpcCtx, md); err != nil {
app.logger.Error("failed to set gRPC header", "err", err)
}
app.logger.Debug("gRPC query received", "type", fmt.Sprintf("%#v", req))
// Catch an OutOfGasPanic caused in the query handlers
defer func() {
if r := recover(); r != nil {
switch rType := r.(type) {
case storetypes.ErrorOutOfGas:
err = errorsmod.Wrapf(sdkerrors.ErrOutOfGas, "Query gas limit exceeded: %v, out of gas in location: %v", sdkCtx.GasMeter().Limit(), rType.Descriptor)
default:
panic(r)
}
}
}()
// set the grpc context back into the SDK context.
// we do this because grpc context has values injected into it that are necessary to retain for systems
// such as OpenTelemetry.
sdkCtx = sdkCtx.WithContext(grpcCtx)
return handler(sdkCtx, req)
}
// Loop through all services and methods, add the interceptor, and register
// the service.
for _, data := range app.GRPCQueryRouter().serviceData {
desc := data.serviceDesc
newMethods := make([]grpc.MethodDesc, len(desc.Methods))
for i, method := range desc.Methods {
methodHandler := method.Handler
newMethods[i] = grpc.MethodDesc{
MethodName: method.MethodName,
Handler: func(srv any, ctx context.Context, dec func(any) error, _ grpc.UnaryServerInterceptor) (any, error) {
return methodHandler(srv, ctx, dec, grpcmiddleware.ChainUnaryServer(
grpcrecovery.UnaryServerInterceptor(),
interceptor,
))
},
}
}
newDesc := &grpc.ServiceDesc{
ServiceName: desc.ServiceName,
HandlerType: desc.HandlerType,
Methods: newMethods,
Streams: desc.Streams,
Metadata: desc.Metadata,
}
server.RegisterService(newDesc, data.handler)
}
}