Add gRPC server & reflection

baseapp/baseapp.go

@@ -294,6 +294,15 @@ func (app *BaseApp) QueryRouter() sdk.QueryRouter { return app.queryRouter }
 // GRPCQueryRouter returns the GRPCQueryRouter of a BaseApp.
 func (app *BaseApp) GRPCQueryRouter() grpc.Server { return app.grpcQueryRouter }
 
+// RegisterGRPC registers gRPC services directly with the gRPC server
+func (app *BaseApp) RegisterGRPC(grpc.Server) {}
+
+// RegisterGRPCProxy registers gRPC services against a proxy that is expected
+// to proxy requests to the ABCI query endpoint
+func (app *BaseApp) RegisterGRPCProxy(server grpc.Server) {
+	app.grpcQueryRouter.RegisterProxyServer(server)
+}
+
 // Seal seals a BaseApp. It prohibits any further modifications to a BaseApp.
 func (app *BaseApp) Seal() { app.sealed = true }
 

baseapp/grpcrouter.go

@@ -16,7 +16,13 @@ var protoCodec = encoding.GetCodec(proto.Name)
 
 // GRPCQueryRouter routes ABCI Query requests to GRPC handlers
 type GRPCQueryRouter struct {
-	routes map[string]GRPCQueryHandler
+	routes      map[string]GRPCQueryHandler
+	serviceData []serviceData
+}
+
+type serviceData struct {
+	serviceDesc *grpc.ServiceDesc
+	handler     interface{}
 }
 
 var _ gogogrpc.Server
@@ -73,4 +79,15 @@ func (qrt *GRPCQueryRouter) RegisterService(sd *grpc.ServiceDesc, handler interf
 			}, nil
 		}
 	}
+
+	qrt.serviceData = append(qrt.serviceData, serviceData{
+		serviceDesc: sd,
+		handler:     handler,
+	})
+}
+
+func (qrt GRPCQueryRouter) RegisterProxyServer(server gogogrpc.Server) {
+	for _, data := range qrt.serviceData {
+		server.RegisterService(data.serviceDesc, data.handler)
+	}
 }

server/config/config.go

@@ -74,13 +74,22 @@ type APIConfig struct {
 	// Ref: https://github.com/cosmos/cosmos-sdk/issues/6420
 }
 
+type GRPCConfig struct {
+	// Enable defines if the gRPC server should be enabled.
+	Enable bool `mapstructure:"enable"`
+
+	// Address defines the API server to listen on
+	Address string `mapstructure:"address"`
+}
+
 // Config defines the server's top level configuration
 type Config struct {
 	BaseConfig `mapstructure:",squash"`
 
 	// Telemetry defines the application telemetry configuration
 	Telemetry telemetry.Config `mapstructure:"telemetry"`
 	API       APIConfig        `mapstructure:"api"`
+	GRPC      GRPCConfig       `mapstructure:"grpc"`
 }
 
 // SetMinGasPrices sets the validator's minimum gas prices.
@@ -129,6 +138,10 @@ func DefaultConfig() *Config {
 			RPCReadTimeout:     10,
 			RPCMaxBodyBytes:    1000000,
 		},
+		GRPC: GRPCConfig{
+			Enable:  true,
+			Address: "tcp://0.0.0.0:9090",
+		},
 	}
 }
 
@@ -161,5 +174,9 @@ func GetConfig() Config {
 			RPCMaxBodyBytes:    viper.GetUint("api.rpc-max-body-bytes"),
 			EnableUnsafeCORS:   viper.GetBool("api.enabled-unsafe-cors"),
 		},
+		GRPC: GRPCConfig{
+			Enable:  viper.GetBool("grpc.enable"),
+			Address: viper.GetString("grpc.address"),
+		},
 	}
 }

server/constructors.go

@@ -6,6 +6,8 @@ import (
 	"os"
 	"path/filepath"
 
+	"github.com/gogo/protobuf/grpc"
+
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/log"
 	tmtypes "github.com/tendermint/tendermint/types"
@@ -23,6 +25,13 @@ type (
 		abci.Application
 
 		RegisterAPIRoutes(*api.Server)
+
+		// RegisterGRPC registers gRPC services directly with the gRPC server
+		RegisterGRPC(grpc.Server)
+
+		// RegisterGRPCProxy registers gRPC services against a proxy that is expected
+		// to proxy requests to the ABCI query endpoint
+		RegisterGRPCProxy(grpc.Server)
 	}
 
 	// AppCreator is a function that allows us to lazily initialize an

server/grpc/proxy.go

@@ -0,0 +1,86 @@
+package grpc
+
+import (
+	"context"
+
+	"github.com/cosmos/cosmos-sdk/client"
+	gogogrpc "github.com/gogo/protobuf/grpc"
+	"github.com/gogo/protobuf/proto"
+	"github.com/tendermint/tendermint/abci/types"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type proxyServer struct {
+	server      *grpc.Server
+	interceptor grpc.UnaryServerInterceptor
+}
+
+func NewProxyServer(server *grpc.Server, interceptor grpc.UnaryServerInterceptor) gogogrpc.Server {
+	return &proxyServer{server: server, interceptor: interceptor}
+}
+
+func (proxy *proxyServer) RegisterService(desc *grpc.ServiceDesc, handler interface{}) {
+	newMethods := make([]grpc.MethodDesc, len(desc.Methods))
+
+	for i, method := range desc.Methods {
+		newMethods[i] = grpc.MethodDesc{
+			MethodName: method.MethodName,
+			Handler: func(srv interface{}, ctx context.Context, dec func(interface{}) error, _ grpc.UnaryServerInterceptor) (interface{}, error) {
+				return method.Handler(srv, ctx, dec, proxy.interceptor)
+			},
+		}
+	}
+
+	newDesc := &grpc.ServiceDesc{
+		ServiceName: desc.ServiceName,
+		HandlerType: desc.HandlerType,
+		Methods:     newMethods,
+		Streams:     desc.Streams,
+		Metadata:    desc.Metadata,
+	}
+
+	proxy.server.RegisterService(newDesc, handler)
+}
+
+func ABCIQueryProxyInterceptor(clientCtx client.Context) grpc.UnaryServerInterceptor {
+	return func(_ context.Context, req interface{}, info *grpc.UnaryServerInfo, _ grpc.UnaryHandler) (resp interface{}, err error) {
+		msg, ok := req.(proto.Message)
+		if !ok {
+			return nil, status.Errorf(codes.Internal, "unable to proto marshal")
+		}
+
+		msgBz, err := proto.Marshal(msg)
+		if err != nil {
+			return nil, err
+		}
+
+		abciReq := types.RequestQuery{
+			Data: msgBz,
+			Path: info.FullMethod,
+		}
+
+		abciRes, err := clientCtx.QueryABCI(abciReq)
+
+		if err != nil {
+			return nil, err
+		}
+
+		if abciRes.Code != 0 {
+			return nil, status.Errorf(codes.Internal, abciRes.Log)
+		}
+
+		respMsg, ok := resp.(proto.Message)
+		if !ok {
+			return nil, status.Errorf(codes.Internal, "unable to proto marshal")
+		}
+
+		err = proto.Unmarshal(abciRes.Value, respMsg)
+		if err != nil {
+			return nil, err
+		}
+
+		return resp, nil
+	}
+}

server/start.go

@@ -4,9 +4,14 @@ package server
 
 import (
 	"fmt"
+	"net"
 	"os"
 	"runtime/pprof"
 
+	"google.golang.org/grpc/reflection"
+
+	"google.golang.org/grpc"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"github.com/tendermint/tendermint/abci/server"
@@ -22,6 +27,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/codec"
 	"github.com/cosmos/cosmos-sdk/server/api"
 	"github.com/cosmos/cosmos-sdk/server/config"
+	grpcproxy "github.com/cosmos/cosmos-sdk/server/grpc"
 )
 
 // Tendermint full-node start flags
@@ -193,27 +199,52 @@ func startInProcess(ctx *Context, cdc codec.JSONMarshaler, appCreator AppCreator
 	}
 
 	config := config.GetConfig()
+
+	genDoc, err := genDocProvider()
+	if err != nil {
+		return err
+	}
+
+	// TODO: Since this is running in process, do we need to provide a verifier
+	// and set TrustNode=false? If so, we need to add additional logic that
+	// waits for a block to be committed first before starting the API server.
+	clientCtx := client.Context{}.
+		WithHomeDir(home).
+		WithChainID(genDoc.ChainID).
+		WithJSONMarshaler(cdc).
+		WithClient(local.New(tmNode)).
+		WithTrustNode(true)
+
 	var apiSrv *api.Server
 	if config.API.Enable {
-		genDoc, err := genDocProvider()
-		if err != nil {
+		apiSrv = api.New(clientCtx)
+		app.RegisterAPIRoutes(apiSrv)
+
+		if err := apiSrv.Start(config); err != nil {
 			return err
 		}
+	}
 
-		// TODO: Since this is running in process, do we need to provide a verifier
-		// and set TrustNode=false? If so, we need to add additional logic that
-		// waits for a block to be committed first before starting the API server.
-		ctx := client.Context{}.
-			WithHomeDir(home).
-			WithChainID(genDoc.ChainID).
-			WithJSONMarshaler(cdc).
-			WithClient(local.New(tmNode)).
-			WithTrustNode(true)
-
-		apiSrv = api.New(ctx)
-		app.RegisterAPIRoutes(apiSrv)
+	var grpcSrv *grpc.Server
+	if config.GRPC.Enable {
+		grpcSrv = grpc.NewServer()
 
-		if err := apiSrv.Start(config); err != nil {
+		app.RegisterGRPC(grpcSrv)
+
+		// proxy queries to the ABCI query endpoint
+		proxyInterceptor := grpcproxy.ABCIQueryProxyInterceptor(clientCtx)
+		proxySrv := grpcproxy.NewProxyServer(grpcSrv, proxyInterceptor)
+		app.RegisterGRPCProxy(proxySrv)
+
+		reflection.Register(grpcSrv)
+
+		listener, err := net.Listen("tcp", config.GRPC.Address)
+		if err != nil {
+			return err
+		}
+
+		err = grpcSrv.Serve(listener)
+		if err != nil {
 			return err
 		}
 	}
@@ -251,6 +282,10 @@ func startInProcess(ctx *Context, cdc codec.JSONMarshaler, appCreator AppCreator
 			_ = apiSrv.Close()
 		}
 
+		if grpcSrv != nil {
+			_ = grpcSrv.Stop
+		}
+
 		ctx.Logger.Info("exiting...")
 	})
 

tests/cli/grpc_query_test.go

@@ -6,6 +6,8 @@ import (
 	"context"
 	"testing"
 
+	"google.golang.org/grpc"
+
 	"github.com/stretchr/testify/require"
 
 	"github.com/cosmos/cosmos-sdk/client"
@@ -26,3 +28,19 @@ func TestCliQueryConn(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "hello", res.Message)
 }
+
+func TestGRPCProxy(t *testing.T) {
+	t.Parallel()
+	f := NewFixtures(t)
+
+	// start simd server
+	proc := f.SDStart()
+	t.Cleanup(func() { proc.Stop(false) })
+
+	conn, err := grpc.Dial("tcp://0.0.0.0:9090")
+	require.NoError(t, err)
+	testClient := testdata.NewTestServiceClient(conn)
+	res, err := testClient.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"})
+	require.NoError(t, err)
+	require.Equal(t, "hello", res.Message)
+}