LaconicNetwork/cosmos-sdk
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

feat(serverv2): integrate gRPC (#21038)

Browse Source 
Co-authored-by: Marko <marko@baricevic.me>
Co-authored-by: marbar3778 <marbar3778@yahoo.com>
This commit is contained in:
testinginprod
2024-07-30 09:26:16 +00:00
committed by GitHub
co-authored by Marko marbar3778
parent c4de9a9706
commit 8b47141655
11 changed files with 190 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/v2/api/grpc/gogoreflection/serverreflection.go
+46 -41
View File
@@ -42,33 +42,41 @@ import (
"errors"
"fmt"
"io"
"log"
"reflect"
"sort"
"strings"
"sync"
//nolint: staticcheck // keep this import for backward compatibility
"github.com/golang/protobuf/proto"
gogoproto "github.com/cosmos/gogoproto/proto"
dpb "github.com/golang/protobuf/protoc-gen-go/descriptor"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/reflect/protodesc"
"google.golang.org/protobuf/reflect/protoreflect"
"cosmossdk.io/core/log"
)
type serverReflectionServer struct {
rpb.UnimplementedServerReflectionServer
s *grpc.Server
methods []string
initSymbols sync.Once
serviceNames []string
symbols map[string]*dpb.FileDescriptorProto // map of fully-qualified names to files
log log.Logger
}
// Register registers the server reflection service on the given gRPC server.
func Register(s *grpc.Server) {
func Register(s *grpc.Server, methods []string, logger log.Logger) {
rpb.RegisterServerReflectionServer(s, &serverReflectionServer{
s: s,
s: s,
methods: methods,
log: logger,
})
}
@@ -82,21 +90,12 @@ type protoMessage interface {
func (s *serverReflectionServer) getSymbols() (svcNames []string, symbolIndex map[string]*dpb.FileDescriptorProto) {
s.initSymbols.Do(func() {
serviceInfo := s.s.GetServiceInfo()
s.symbols = map[string]*dpb.FileDescriptorProto{}
s.serviceNames = make([]string, 0, len(serviceInfo))
services, fds := s.getServices(s.methods)
s.serviceNames = services
processed := map[string]struct{}{}
for svc, info := range serviceInfo {
s.serviceNames = append(s.serviceNames, svc)
fdenc, ok := parseMetadata(info.Metadata)
if !ok {
continue
}
fd, err := decodeFileDesc(fdenc)
if err != nil {
continue
}
for _, fd := range fds {
s.processFile(fd, processed)
}
sort.Strings(s.serviceNames)
@@ -207,7 +206,7 @@ func decodeFileDesc(enc []byte) (*dpb.FileDescriptorProto, error) {
}
fd := new(dpb.FileDescriptorProto)
if err := proto.Unmarshal(raw, fd); err != nil {
if err := gogoproto.Unmarshal(raw, fd); err != nil {
return nil, fmt.Errorf("bad descriptor: %w", err)
}
return fd, nil
@@ -237,7 +236,7 @@ func typeForName(name string) (reflect.Type, error) {
}
func fileDescContainingExtension(st reflect.Type, ext int32) (*dpb.FileDescriptorProto, error) {
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(proto.Message)
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(gogoproto.Message)
if !ok {
return nil, fmt.Errorf("failed to create message from type: %v", st)
}
@@ -252,7 +251,7 @@ func fileDescContainingExtension(st reflect.Type, ext int32) (*dpb.FileDescripto
}
func (s *serverReflectionServer) allExtensionNumbersForType(st reflect.Type) ([]int32, error) {
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(proto.Message)
m, ok := reflect.Zero(reflect.PtrTo(st)).Interface().(gogoproto.Message)
if !ok {
return nil, fmt.Errorf("failed to create message from type: %v", st)
}
@@ -272,7 +271,7 @@ func fileDescWithDependencies(fd *dpb.FileDescriptorProto, sentFileDescriptors m
queue = queue[1:]
if sent := sentFileDescriptors[currentfd.GetName()]; len(r) == 0 || !sent {
sentFileDescriptors[currentfd.GetName()] = true
currentfdEncoded, err := proto.Marshal(currentfd)
currentfdEncoded, err := gogoproto.Marshal(currentfd)
if err != nil {
return nil, err
}
@@ -305,24 +304,6 @@ func (s *serverReflectionServer) fileDescEncodingByFilename(name string, sentFil
return fileDescWithDependencies(fd, sentFileDescriptors)
}
// parseMetadata finds the file descriptor bytes specified meta.
// For SupportPackageIsVersion4, m is the name of the proto file, we
// call proto.FileDescriptor to get the byte slice.
// For SupportPackageIsVersion3, m is a byte slice itself.
func parseMetadata(meta interface{}) ([]byte, bool) {
// Check if meta is the file name.
if fileNameForMeta, ok := meta.(string); ok {
return getFileDescriptor(fileNameForMeta), true
}
// Check if meta is the byte slice.
if enc, ok := meta.([]byte); ok {
return enc, true
}
return nil, false
}
// fileDescEncodingContainingSymbol finds the file descriptor containing the
// given symbol, finds all of its previously unsent transitive dependencies,
// does marshaling on them, and returns the marshaled result. The given symbol
@@ -446,7 +427,6 @@ func (s *serverReflectionServer) ServerReflectionInfo(stream rpb.ServerReflectio
ErrorMessage: err.Error(),
},
}
log.Printf("OH NO: %s", err)
} else {
out.MessageResponse = &rpb.ServerReflectionResponse_AllExtensionNumbersResponse{
AllExtensionNumbersResponse: &rpb.ExtensionNumberResponse{ //nolint:staticcheck // SA1019: we want to keep using v1alpha
@@ -476,3 +456,28 @@ func (s *serverReflectionServer) ServerReflectionInfo(stream rpb.ServerReflectio
}
}
}
// getServices gets the unique list of services given a list of methods.
func (s *serverReflectionServer) getServices(methods []string) (svcs []string, fds []*dpb.FileDescriptorProto) {
registry, err := gogoproto.MergedRegistry()
if err != nil {
s.log.Error("unable to load merged registry", "err", err)
return nil, nil
}
seenSvc := map[protoreflect.FullName]struct{}{}
for _, methodName := range methods {
methodName = strings.Join(strings.Split(methodName[1:], "/"), ".")
md, err := registry.FindDescriptorByName(protoreflect.FullName(methodName))
if err != nil {
s.log.Error("unable to load method descriptor", "method", methodName, "err", err)
continue
}
svc := md.(protoreflect.MethodDescriptor).Parent()
if _, seen := seenSvc[svc.FullName()]; !seen {
svcs = append(svcs, string(svc.FullName()))
file := svc.ParentFile()
fds = append(fds, protodesc.ToFileDescriptorProto(file))
}
}
return
}
server/v2/api/grpc/server.go
+104 -11
View File
@@ -2,11 +2,20 @@ package grpc
import (
"context"
"errors"
"fmt"
"io"
"net"
"strconv"
"github.com/cosmos/gogoproto/proto"
"github.com/spf13/pflag"
"github.com/spf13/viper"
"golang.org/x/exp/maps"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"cosmossdk.io/core/transaction"
"cosmossdk.io/log"
@@ -14,7 +23,12 @@ import (
"cosmossdk.io/server/v2/api/grpc/gogoreflection"
)
type GRPCServer[T transaction.Tx] struct {
const (
BlockHeightHeader = "x-cosmos-block-height"
FlagAddress = "address"
)
type Server[T transaction.Tx] struct {
logger log.Logger
config *Config
cfgOptions []CfgOption
@@ -23,32 +37,34 @@ type GRPCServer[T transaction.Tx] struct {
}
// New creates a new grpc server.
func New[T transaction.Tx](cfgOptions ...CfgOption) *GRPCServer[T] {
return &GRPCServer[T]{
func New[T transaction.Tx](cfgOptions ...CfgOption) *Server[T] {
return &Server[T]{
cfgOptions: cfgOptions,
}
}
// Init returns a correctly configured and initialized gRPC server.
// Note, the caller is responsible for starting the server.
func (s *GRPCServer[T]) Init(appI serverv2.AppI[T], v *viper.Viper, logger log.Logger) error {
func (s *Server[T]) Init(appI serverv2.AppI[T], v *viper.Viper, logger log.Logger) error {
cfg := s.Config().(*Config)
if v != nil {
if err := serverv2.UnmarshalSubConfig(v, s.Name(), &cfg); err != nil {
return fmt.Errorf("failed to unmarshal config: %w", err)
}
}
methodsMap := appI.GetGPRCMethodsToMessageMap()
grpcSrv := grpc.NewServer(
grpc.ForceServerCodec(newProtoCodec(appI.InterfaceRegistry()).GRPCCodec()),
grpc.MaxSendMsgSize(cfg.MaxSendMsgSize),
grpc.MaxRecvMsgSize(cfg.MaxRecvMsgSize),
grpc.UnknownServiceHandler(
makeUnknownServiceHandler(methodsMap, appI.GetAppManager()),
),
)
// appI.RegisterGRPCServer(grpcSrv)
// Reflection allows external clients to see what services and methods the gRPC server exposes.
gogoreflection.Register(grpcSrv)
gogoreflection.Register(grpcSrv, maps.Keys(methodsMap), logger.With("sub-module", "grpc-reflection"))
s.grpcSrv = grpcSrv
s.config = cfg
@@ -57,11 +73,88 @@ func (s *GRPCServer[T]) Init(appI serverv2.AppI[T], v *viper.Viper, logger log.L
return nil
}
func (s *GRPCServer[T]) Name() string {
func (s *Server[T]) StartCmdFlags() *pflag.FlagSet {
flags := pflag.NewFlagSet("grpc", pflag.ExitOnError)
// start flags are prefixed with the server name
// as the config in prefixed with the server name
// this allows viper to properly bind the flags
prefix := func(f string) string {
return fmt.Sprintf("%s.%s", s.Name(), f)
}
flags.String(prefix(FlagAddress), "localhost:9090", "Listen address")
return flags
}
func makeUnknownServiceHandler(messageMap map[string]func() proto.Message, querier interface {
Query(ctx context.Context, version uint64, msg proto.Message) (proto.Message, error)
},
) grpc.StreamHandler {
return func(srv any, stream grpc.ServerStream) error {
method, ok := grpc.MethodFromServerStream(stream)
if !ok {
return status.Error(codes.InvalidArgument, "unable to get method")
}
makeMsg, exists := messageMap[method]
if !exists {
return status.Errorf(codes.Unimplemented, "gRPC method %s is not handled", method)
}
for {
req := makeMsg()
err := stream.RecvMsg(req)
if err != nil {
if errors.Is(err, io.EOF) {
return nil
}
return err
}
// extract height header
ctx := stream.Context()
height, err := getHeightFromCtx(ctx)
if err != nil {
return status.Errorf(codes.InvalidArgument, "invalid get height from context: %v", err)
}
resp, err := querier.Query(ctx, height, req)
if err != nil {
return err
}
err = stream.SendMsg(resp)
if err != nil {
return err
}
}
}
}
func getHeightFromCtx(ctx context.Context) (uint64, error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return 0, nil
}
values := md.Get(BlockHeightHeader)
if len(values) == 0 {
return 0, nil
}
if len(values) != 1 {
return 0, fmt.Errorf("gRPC height metadata must be of length 1, got: %d", len(values))
}
heightStr := values[0]
height, err := strconv.ParseUint(heightStr, 10, 64)
if err != nil {
return 0, fmt.Errorf("unable to parse height string from gRPC metadata %s: %w", heightStr, err)
}
return height, nil
}
func (s *Server[T]) Name() string {
return "grpc"
}
func (s *GRPCServer[T]) Config() any {
func (s *Server[T]) Config() any {
if s.config == nil || s.config == (&Config{}) {
cfg := DefaultConfig()
// overwrite the default config with the provided options
@@ -75,7 +168,7 @@ func (s *GRPCServer[T]) Config() any {
return s.config
}
func (s *GRPCServer[T]) Start(ctx context.Context) error {
func (s *Server[T]) Start(ctx context.Context) error {
if !s.config.Enable {
return nil
}
@@ -102,7 +195,7 @@ func (s *GRPCServer[T]) Start(ctx context.Context) error {
return err
}
func (s *GRPCServer[T]) Stop(ctx context.Context) error {
func (s *Server[T]) Stop(ctx context.Context) error {
if !s.config.Enable {
return nil
}