From ca5dc0e3c33f8a2d53ae389110e7f71a07235317 Mon Sep 17 00:00:00 2001 From: jesse snyder Date: Mon, 27 Mar 2023 13:30:46 -0600 Subject: [PATCH] Feature/grpc execution api (#1) * add buf config and generated grpc code * poc e2e grpc communication * comment out panicking code * logging help * now reads cli args. now stops grpc server on shutdown. * add mutex to GRPCServerHandler --- cmd/geth/config.go | 5 + cmd/geth/main.go | 3 + cmd/utils/flags.go | 40 +++ go.mod | 2 + go.sum | 4 + grpc/README.md | 15 ++ grpc/buf.gen.yaml | 22 ++ grpc/execution/server.go | 64 +++++ grpc/gen/proto/execution/v1/execution.pb.go | 236 ++++++++++++++++++ .../proto/execution/v1/execution_grpc.pb.go | 109 ++++++++ node/config.go | 23 ++ node/defaults.go | 6 + node/grpcstack.go | 75 ++++++ node/node.go | 44 ++++ 14 files changed, 648 insertions(+) create mode 100644 grpc/README.md create mode 100644 grpc/buf.gen.yaml create mode 100644 grpc/execution/server.go create mode 100644 grpc/gen/proto/execution/v1/execution.pb.go create mode 100644 grpc/gen/proto/execution/v1/execution_grpc.pb.go create mode 100644 node/grpcstack.go diff --git a/cmd/geth/config.go b/cmd/geth/config.go index 0b856d1c1b..75becbf590 100644 --- a/cmd/geth/config.go +++ b/cmd/geth/config.go @@ -172,6 +172,11 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) { utils.RegisterGraphQLService(stack, backend, filterSystem, &cfg.Node) } + // Configure gRPC if requested. + if ctx.IsSet(utils.GRPCEnabledFlag.Name) { + utils.RegisterGRPCService(stack, backend, &cfg.Node) + } + // Add the Ethereum Stats daemon if requested. if cfg.Ethstats.URL != "" { utils.RegisterEthStatsService(stack, backend, cfg.Ethstats.URL) diff --git a/cmd/geth/main.go b/cmd/geth/main.go index a970e76523..03c9f29157 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -180,6 +180,9 @@ var ( utils.RPCGlobalEVMTimeoutFlag, utils.RPCGlobalTxFeeCapFlag, utils.AllowUnprotectedTxs, + utils.GRPCEnabledFlag, + utils.GRPCHostFlag, + utils.GRPCPortFlag, } metricsFlags = []cli.Flag{ diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 08de71ee83..02bb9f2672 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -791,6 +791,24 @@ var ( Usage: "Enables the (deprecated) personal namespace", Category: flags.APICategory, } + // grpc + GRPCEnabledFlag = &cli.BoolFlag{ + Name: "grpc", + Usage: "Enable the gRPC server", + Category: flags.APICategory, + } + GRPCHostFlag = &cli.StringFlag{ + Name: "grpc.addr", + Usage: "gRPC server listening interface", + Value: node.DefaultGRPCHost, + Category: flags.APICategory, + } + GRPCPortFlag = &cli.IntFlag{ + Name: "grpc.port", + Usage: "gRPC server listening port", + Value: node.DefaultGRPCPort, + Category: flags.APICategory, + } // Network Settings MaxPeersFlag = &cli.IntFlag{ @@ -1211,6 +1229,19 @@ func setHTTP(ctx *cli.Context, cfg *node.Config) { } } +// setGRCP creates the gRPC RPC listener interface string from the set command +// line flags, returning empty if the gRPC endpoint is disabled. +func setGRCP(ctx *cli.Context, cfg *node.Config) { + if ctx.Bool(GRPCEnabledFlag.Name) { + if ctx.IsSet(GRPCHostFlag.Name) { + cfg.GRPCHost = ctx.String(GRPCHostFlag.Name) + } + if ctx.IsSet(GRPCPortFlag.Name) { + cfg.GRPCPort = ctx.Int(GRPCPortFlag.Name) + } + } +} + // setGraphQL creates the GraphQL listener interface string from the set // command line flags, returning empty if the GraphQL endpoint is disabled. func setGraphQL(ctx *cli.Context, cfg *node.Config) { @@ -1460,6 +1491,7 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) { SetP2PConfig(ctx, &cfg.P2P) setIPC(ctx, cfg) setHTTP(ctx, cfg) + setGRCP(ctx, cfg) setGraphQL(ctx, cfg) setWS(ctx, cfg) setNodeUserIdent(ctx, cfg) @@ -2032,6 +2064,14 @@ func RegisterGraphQLService(stack *node.Node, backend ethapi.Backend, filterSyst } } +// RegisterGRPCService adds the gRPC API to the node. +// It was done this way so that our grpc execution server can access the ethapi.Backend +func RegisterGRPCService(stack *node.Node, backend ethapi.Backend, cfg *node.Config) { + if err := node.NewGRPCServerHandler(stack, backend, cfg); err != nil { + Fatalf("Failed to register the gRPC service: %v", err) + } +} + // RegisterFilterAPI adds the eth log filtering RPC API to the node. func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconfig.Config) *filters.FilterSystem { isLightClient := ethcfg.SyncMode == downloader.LightSync diff --git a/go.mod b/go.mod index 95c80c6fd8..7cefbb9121 100644 --- a/go.mod +++ b/go.mod @@ -119,6 +119,8 @@ require ( golang.org/x/mod v0.9.0 // indirect golang.org/x/net v0.8.0 // indirect golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect + google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect + google.golang.org/grpc v1.53.0 // indirect google.golang.org/protobuf v1.28.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 4c134a2772..a24c2066e3 100644 --- a/go.sum +++ b/go.sum @@ -586,12 +586,16 @@ google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoA google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc= google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= google.golang.org/genproto v0.0.0-20210624195500-8bfb893ecb84/go.mod h1:SzzZ/N+nwJDaO1kznhnlzqS8ocJICar6hYhVyhi++24= +google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f h1:BWUVssLB0HVOSY78gIdvk1dTVYtT1y8SBWtPYuTJ/6w= +google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f/go.mod h1:RGgjbofJ8xD9Sq1VVhDM1Vok1vRONV+rg+CjzG4SZKM= google.golang.org/grpc v1.12.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg= google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY= google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk= google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM= +google.golang.org/grpc v1.53.0 h1:LAv2ds7cmFV/XTS3XG1NneeENYrXGmorPxsBbptIjNc= +google.golang.org/grpc v1.53.0/go.mod h1:OnIrk0ipVdj4N5d9IUoFUx72/VlD7+jUsHwZgwSMQpw= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/grpc/README.md b/grpc/README.md new file mode 100644 index 0000000000..75af561f68 --- /dev/null +++ b/grpc/README.md @@ -0,0 +1,15 @@ +This package provides a gRPC server as an entrypoint to the EVM. + +Helpful commands (MacOS): +```bash +# install necessary dependencies +brew install leveldb + +# build geth +make geth + +# TODO - run beacon? + +# run geth +./build/bin/geth --grpc --grpc.addr "[::1]" --grpc.port 50051 +``` diff --git a/grpc/buf.gen.yaml b/grpc/buf.gen.yaml new file mode 100644 index 0000000000..32ceca95f0 --- /dev/null +++ b/grpc/buf.gen.yaml @@ -0,0 +1,22 @@ +# buf.gen.yaml +version: v1 +managed: + enabled: true + go_package_prefix: + # : name in go.mod + # : where generated code should be output + # default: / + default: github.com/ethereum/go-ethereum/grpc + # Remove `except` field if googleapis is not used + # except: + # - buf.build/googleapis/googleapis +plugins: + - plugin: buf.build/grpc/go + out: gen + opt: + - paths=source_relative + # dependencies + - plugin: buf.build/protocolbuffers/go + out: gen + opt: + - paths=source_relative diff --git a/grpc/execution/server.go b/grpc/execution/server.go new file mode 100644 index 0000000000..5c32c51280 --- /dev/null +++ b/grpc/execution/server.go @@ -0,0 +1,64 @@ +// Package execution provides the gRPC server for the execution layer. +// +// Its procedures will be called from the conductor. It is responsible +// for immediately executing lists of ordered transactions that come from the shared sequencer. +package execution + +import ( + "context" + + "github.com/ethereum/go-ethereum/core/types" + executionv1 "github.com/ethereum/go-ethereum/grpc/gen/proto/execution/v1" + "github.com/ethereum/go-ethereum/internal/ethapi" + "github.com/ethereum/go-ethereum/log" +) + +// executionServiceServer is the implementation of the ExecutionServiceServer interface. +type ExecutionServiceServer struct { + // NOTE - from the generated code: + // All implementations must embed UnimplementedExecutionServiceServer + // for forward compatibility + executionv1.UnimplementedExecutionServiceServer + + // TODO - will need access to the consensus api to call functions for building a block + // e.g. getPayload, newPayload, forkchoiceUpdated + + Backend ethapi.Backend + + // TODO - will need access to forkchoice on first run. + // this will probably be passed in when calling NewServer +} + +// FIXME - how do we know which hash to start with? will probably need another api function like +// GetHeadHash() to get the head hash of the forkchoice + +func (s *ExecutionServiceServer) DoBlock(ctx context.Context, req *executionv1.DoBlockRequest) (*executionv1.DoBlockResponse, error) { + log.Info("DoBlock called request", "request", req) + + // NOTE - Request.Header.ParentHash needs to match forkchoice head hash + // ParentHash should be the forkchoice head of the last block + + // TODO - need to call consensus api to build a block + + // txs := bytesToTransactions(req.Transactions) + // for _, tx := range txs { + // s.Backend.SendTx(ctx, tx) + // } + + res := &executionv1.DoBlockResponse{ + // TODO - get state root from last block + StateRoot: []byte{0x00}, + } + return res, nil +} + +// convert bytes to transactions +func bytesToTransactions(b [][]byte) []*types.Transaction { + txs := []*types.Transaction{} + for _, txBytes := range b { + tx := &types.Transaction{} + tx.UnmarshalBinary(txBytes) + txs = append(txs, tx) + } + return txs +} diff --git a/grpc/gen/proto/execution/v1/execution.pb.go b/grpc/gen/proto/execution/v1/execution.pb.go new file mode 100644 index 0000000000..aca5c12348 --- /dev/null +++ b/grpc/gen/proto/execution/v1/execution.pb.go @@ -0,0 +1,236 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.30.0 +// protoc (unknown) +// source: proto/execution/v1/execution.proto + +package executionv1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type DoBlockRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Header []byte `protobuf:"bytes,1,opt,name=header,proto3" json:"header,omitempty"` + Transactions [][]byte `protobuf:"bytes,2,rep,name=transactions,proto3" json:"transactions,omitempty"` +} + +func (x *DoBlockRequest) Reset() { + *x = DoBlockRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_execution_v1_execution_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *DoBlockRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DoBlockRequest) ProtoMessage() {} + +func (x *DoBlockRequest) ProtoReflect() protoreflect.Message { + mi := &file_proto_execution_v1_execution_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DoBlockRequest.ProtoReflect.Descriptor instead. +func (*DoBlockRequest) Descriptor() ([]byte, []int) { + return file_proto_execution_v1_execution_proto_rawDescGZIP(), []int{0} +} + +func (x *DoBlockRequest) GetHeader() []byte { + if x != nil { + return x.Header + } + return nil +} + +func (x *DoBlockRequest) GetTransactions() [][]byte { + if x != nil { + return x.Transactions + } + return nil +} + +type DoBlockResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + StateRoot []byte `protobuf:"bytes,1,opt,name=state_root,json=stateRoot,proto3" json:"state_root,omitempty"` +} + +func (x *DoBlockResponse) Reset() { + *x = DoBlockResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_proto_execution_v1_execution_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *DoBlockResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DoBlockResponse) ProtoMessage() {} + +func (x *DoBlockResponse) ProtoReflect() protoreflect.Message { + mi := &file_proto_execution_v1_execution_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DoBlockResponse.ProtoReflect.Descriptor instead. +func (*DoBlockResponse) Descriptor() ([]byte, []int) { + return file_proto_execution_v1_execution_proto_rawDescGZIP(), []int{1} +} + +func (x *DoBlockResponse) GetStateRoot() []byte { + if x != nil { + return x.StateRoot + } + return nil +} + +var File_proto_execution_v1_execution_proto protoreflect.FileDescriptor + +var file_proto_execution_v1_execution_proto_rawDesc = []byte{ + 0x0a, 0x22, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, + 0x6e, 0x2f, 0x76, 0x31, 0x2f, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2e, + 0x76, 0x31, 0x22, 0x4c, 0x0a, 0x0e, 0x44, 0x6f, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x22, 0x0a, 0x0c, + 0x74, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x02, 0x20, 0x03, + 0x28, 0x0c, 0x52, 0x0c, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x22, 0x30, 0x0a, 0x0f, 0x44, 0x6f, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, + 0x6e, 0x73, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x73, 0x74, 0x61, 0x74, 0x65, 0x5f, 0x72, 0x6f, 0x6f, + 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x73, 0x74, 0x61, 0x74, 0x65, 0x52, 0x6f, + 0x6f, 0x74, 0x32, 0x5a, 0x0a, 0x10, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x53, + 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x46, 0x0a, 0x07, 0x44, 0x6f, 0x42, 0x6c, 0x6f, 0x63, + 0x6b, 0x12, 0x1c, 0x2e, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x76, 0x31, + 0x2e, 0x44, 0x6f, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, + 0x1d, 0x2e, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x44, + 0x6f, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0xb8, + 0x01, 0x0a, 0x10, 0x63, 0x6f, 0x6d, 0x2e, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, + 0x2e, 0x76, 0x31, 0x42, 0x0e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, + 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x43, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, + 0x6d, 0x2f, 0x65, 0x74, 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2f, 0x67, 0x6f, 0x2d, 0x65, 0x74, + 0x68, 0x65, 0x72, 0x65, 0x75, 0x6d, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x70, 0x72, 0x6f, 0x74, + 0x6f, 0x2f, 0x65, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2f, 0x76, 0x31, 0x3b, 0x65, + 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x76, 0x31, 0xa2, 0x02, 0x03, 0x45, 0x58, 0x58, + 0xaa, 0x02, 0x0c, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x56, 0x31, 0xca, + 0x02, 0x0c, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x5c, 0x56, 0x31, 0xe2, 0x02, + 0x18, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, + 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0d, 0x45, 0x78, 0x65, 0x63, + 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x33, +} + +var ( + file_proto_execution_v1_execution_proto_rawDescOnce sync.Once + file_proto_execution_v1_execution_proto_rawDescData = file_proto_execution_v1_execution_proto_rawDesc +) + +func file_proto_execution_v1_execution_proto_rawDescGZIP() []byte { + file_proto_execution_v1_execution_proto_rawDescOnce.Do(func() { + file_proto_execution_v1_execution_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_execution_v1_execution_proto_rawDescData) + }) + return file_proto_execution_v1_execution_proto_rawDescData +} + +var file_proto_execution_v1_execution_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_proto_execution_v1_execution_proto_goTypes = []interface{}{ + (*DoBlockRequest)(nil), // 0: execution.v1.DoBlockRequest + (*DoBlockResponse)(nil), // 1: execution.v1.DoBlockResponse +} +var file_proto_execution_v1_execution_proto_depIdxs = []int32{ + 0, // 0: execution.v1.ExecutionService.DoBlock:input_type -> execution.v1.DoBlockRequest + 1, // 1: execution.v1.ExecutionService.DoBlock:output_type -> execution.v1.DoBlockResponse + 1, // [1:2] is the sub-list for method output_type + 0, // [0:1] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_proto_execution_v1_execution_proto_init() } +func file_proto_execution_v1_execution_proto_init() { + if File_proto_execution_v1_execution_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_proto_execution_v1_execution_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*DoBlockRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_proto_execution_v1_execution_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*DoBlockResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_proto_execution_v1_execution_proto_rawDesc, + NumEnums: 0, + NumMessages: 2, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_proto_execution_v1_execution_proto_goTypes, + DependencyIndexes: file_proto_execution_v1_execution_proto_depIdxs, + MessageInfos: file_proto_execution_v1_execution_proto_msgTypes, + }.Build() + File_proto_execution_v1_execution_proto = out.File + file_proto_execution_v1_execution_proto_rawDesc = nil + file_proto_execution_v1_execution_proto_goTypes = nil + file_proto_execution_v1_execution_proto_depIdxs = nil +} diff --git a/grpc/gen/proto/execution/v1/execution_grpc.pb.go b/grpc/gen/proto/execution/v1/execution_grpc.pb.go new file mode 100644 index 0000000000..63995f7e87 --- /dev/null +++ b/grpc/gen/proto/execution/v1/execution_grpc.pb.go @@ -0,0 +1,109 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.3.0 +// - protoc (unknown) +// source: proto/execution/v1/execution.proto + +package executionv1 + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.32.0 or later. +const _ = grpc.SupportPackageIsVersion7 + +const ( + ExecutionService_DoBlock_FullMethodName = "/execution.v1.ExecutionService/DoBlock" +) + +// ExecutionServiceClient is the client API for ExecutionService service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type ExecutionServiceClient interface { + DoBlock(ctx context.Context, in *DoBlockRequest, opts ...grpc.CallOption) (*DoBlockResponse, error) +} + +type executionServiceClient struct { + cc grpc.ClientConnInterface +} + +func NewExecutionServiceClient(cc grpc.ClientConnInterface) ExecutionServiceClient { + return &executionServiceClient{cc} +} + +func (c *executionServiceClient) DoBlock(ctx context.Context, in *DoBlockRequest, opts ...grpc.CallOption) (*DoBlockResponse, error) { + out := new(DoBlockResponse) + err := c.cc.Invoke(ctx, ExecutionService_DoBlock_FullMethodName, in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// ExecutionServiceServer is the server API for ExecutionService service. +// All implementations must embed UnimplementedExecutionServiceServer +// for forward compatibility +type ExecutionServiceServer interface { + DoBlock(context.Context, *DoBlockRequest) (*DoBlockResponse, error) + mustEmbedUnimplementedExecutionServiceServer() +} + +// UnimplementedExecutionServiceServer must be embedded to have forward compatible implementations. +type UnimplementedExecutionServiceServer struct { +} + +func (UnimplementedExecutionServiceServer) DoBlock(context.Context, *DoBlockRequest) (*DoBlockResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method DoBlock not implemented") +} +func (UnimplementedExecutionServiceServer) mustEmbedUnimplementedExecutionServiceServer() {} + +// UnsafeExecutionServiceServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to ExecutionServiceServer will +// result in compilation errors. +type UnsafeExecutionServiceServer interface { + mustEmbedUnimplementedExecutionServiceServer() +} + +func RegisterExecutionServiceServer(s grpc.ServiceRegistrar, srv ExecutionServiceServer) { + s.RegisterService(&ExecutionService_ServiceDesc, srv) +} + +func _ExecutionService_DoBlock_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(DoBlockRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ExecutionServiceServer).DoBlock(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ExecutionService_DoBlock_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ExecutionServiceServer).DoBlock(ctx, req.(*DoBlockRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// ExecutionService_ServiceDesc is the grpc.ServiceDesc for ExecutionService service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var ExecutionService_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "execution.v1.ExecutionService", + HandlerType: (*ExecutionServiceServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "DoBlock", + Handler: _ExecutionService_DoBlock_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "proto/execution/v1/execution.proto", +} diff --git a/node/config.go b/node/config.go index 37a7d5837f..6be260efbf 100644 --- a/node/config.go +++ b/node/config.go @@ -189,6 +189,12 @@ type Config struct { // Requests using ip address directly are not affected GraphQLVirtualHosts []string `toml:",omitempty"` + // GRPCHost is the host interface on which to start the gRPC server. If this + // field is empty, no gRPC API endpoint will be started. + GRPCHost string `toml:",omitempty"` + // GRPCPort is the TCP port number on which to start the gRPC server. + GRPCPort int `toml:",omitempty"` + // Logger is a custom logger to use with the p2p.Server. Logger log.Logger `toml:",omitempty"` @@ -260,12 +266,29 @@ func (c *Config) HTTPEndpoint() string { return fmt.Sprintf("%s:%d", c.HTTPHost, c.HTTPPort) } +// GRPCEndpoint resolves a gRPC endpoint based on the configured host interface +// and port parameters. +func (c *Config) GRPCEndpoint() string { + if c.GRPCHost == "" { + return "" + } + return fmt.Sprintf("%s:%d", c.GRPCHost, c.GRPCPort) +} + // DefaultHTTPEndpoint returns the HTTP endpoint used by default. func DefaultHTTPEndpoint() string { config := &Config{HTTPHost: DefaultHTTPHost, HTTPPort: DefaultHTTPPort, AuthPort: DefaultAuthPort} return config.HTTPEndpoint() } +// DefaultGRPCEndpoint returns the gRPC endpoint used by default. +// NOTE - implemented this to be consistent with DefaultHTTPEndpoint, but +// neither are ever used +func DefaultGRPCEndpoint() string { + config := &Config{GRPCHost: DefaultGRPCHost, GRPCPort: DefaultGRPCPort} + return config.GRPCEndpoint() +} + // WSEndpoint resolves a websocket endpoint based on the configured host interface // and port parameters. func (c *Config) WSEndpoint() string { diff --git a/node/defaults.go b/node/defaults.go index 96ebed81c5..42c339a0c9 100644 --- a/node/defaults.go +++ b/node/defaults.go @@ -36,6 +36,9 @@ const ( DefaultGraphQLPort = 8547 // Default TCP port for the GraphQL server DefaultAuthHost = "localhost" // Default host interface for the authenticated apis DefaultAuthPort = 8551 // Default port for the authenticated apis + // grpc + DefaultGRPCHost = "[::1]" // Default host interface for the gRPC server + DefaultGRPCPort = 50051 // Default port for the gRPC server ) var ( @@ -65,6 +68,9 @@ var DefaultConfig = Config{ NAT: nat.Any(), }, DBEngine: "", + // grpc + GRPCHost: DefaultGRPCHost, + GRPCPort: DefaultGRPCPort, } // DefaultDataDir is the default data directory to use for the databases and other diff --git a/node/grpcstack.go b/node/grpcstack.go new file mode 100644 index 0000000000..1c0c3f4768 --- /dev/null +++ b/node/grpcstack.go @@ -0,0 +1,75 @@ +package node + +import ( + "net" + "sync" + + "github.com/ethereum/go-ethereum/grpc/execution" + executionv1 "github.com/ethereum/go-ethereum/grpc/gen/proto/execution/v1" + "github.com/ethereum/go-ethereum/internal/ethapi" + "github.com/ethereum/go-ethereum/log" + "google.golang.org/grpc" +) + +// GRPCServerHandler is the gRPC server handler. +// It gives us a way to attach the gRPC server to the node so it can be stopped on shutdown. +type GRPCServerHandler struct { + mu sync.Mutex + + endpoint string + server *grpc.Server + executionServiceServer *execution.ExecutionServiceServer +} + +// NewServer creates a new gRPC server. +// It registers the execution service server. +// It registers the gRPC server with the node so it can be stopped on shutdown. +func NewGRPCServerHandler(node *Node, backend ethapi.Backend, cfg *Config) error { + server := grpc.NewServer() + + executionServiceServer := &execution.ExecutionServiceServer{ + Backend: backend, + } + + log.Info("gRPC server enabled", "endpoint", cfg.GRPCEndpoint()) + + serverHandler := &GRPCServerHandler{ + endpoint: cfg.GRPCEndpoint(), + server: server, + executionServiceServer: executionServiceServer, + } + + executionv1.RegisterExecutionServiceServer(server, executionServiceServer) + + node.RegisterGRPCServer(serverHandler) + return nil +} + +// Start starts the gRPC server if it is enabled. +func (handler *GRPCServerHandler) Start() error { + handler.mu.Lock() + defer handler.mu.Unlock() + + if handler.endpoint == "" { + return nil + } + + // Start the gRPC server + lis, err := net.Listen("tcp", handler.endpoint) + if err != nil { + return err + } + go handler.server.Serve(lis) + log.Info("gRPC server started", "endpoint", handler.endpoint) + return nil +} + +// Stop stops the gRPC server. +func (handler *GRPCServerHandler) Stop() error { + handler.mu.Lock() + defer handler.mu.Unlock() + + handler.server.Stop() + log.Info("gRPC server stopped", "endpoint", handler.endpoint) + return nil +} diff --git a/node/node.go b/node/node.go index 2f89bc1ad2..e13dd2f8e3 100644 --- a/node/node.go +++ b/node/node.go @@ -64,6 +64,9 @@ type Node struct { ipc *ipcServer // Stores information about the ipc http server inprocHandler *rpc.Server // In-process RPC request handler to process the API requests + // grpc + grpcServerHandler *GRPCServerHandler // Stores information about the grpc server + databases map[*closeTrackingDB]struct{} // All open databases } @@ -274,6 +277,11 @@ func (n *Node) openEndpoints() error { n.stopRPC() n.server.Stop() } + // start GRPC endpoints + err = n.startGRPC() + if err != nil { + n.stopGRPC() + } return err } @@ -303,6 +311,9 @@ func (n *Node) stopServices(running []Lifecycle) error { // Stop p2p networking. n.server.Stop() + // Stop GRPC server + n.stopGRPC() + if len(failure.Services) > 0 { return failure } @@ -521,6 +532,21 @@ func (n *Node) stopRPC() { n.stopInProc() } +func (n *Node) startGRPC() error { + if n.config.GRPCHost != "" { + // start the server + if err := n.grpcServerHandler.Start(); err != nil { + return err + } + } + + return nil +} + +func (n *Node) stopGRPC() { + n.grpcServerHandler.Stop() +} + // startInProc registers all RPC APIs on the inproc server. func (n *Node) startInProc(apis []rpc.API) error { for _, api := range apis { @@ -588,6 +614,19 @@ func (n *Node) getAPIs() (unauthenticated, all []rpc.API) { return unauthenticated, n.rpcAPIs } +// RegisterGRPCServer registers a gRPC server on the node. +// This allows us to control grpc server startup and shutdown from the node. +func (n *Node) RegisterGRPCServer(handler *GRPCServerHandler) { + n.lock.Lock() + defer n.lock.Unlock() + + if n.state != initializingState { + panic("can't register gRPC server on running/stopped node") + } + + n.grpcServerHandler = handler +} + // RegisterHandler mounts a handler on the given path on the canonical HTTP server. // // The name of the handler is shown in a log message when the HTTP server starts @@ -667,6 +706,11 @@ func (n *Node) HTTPEndpoint() string { return "http://" + n.http.listenAddr() } +// GRPCENDPOINT returns the URL of the GRPC server. +func (n *Node) GRPCEndpoint() string { + return "http://" + n.grpcServerHandler.endpoint +} + // WSEndpoint returns the current JSON-RPC over WebSocket endpoint. func (n *Node) WSEndpoint() string { if n.http.wsAllowed() {