mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2025-01-15 22:48:19 +00:00
0326be86b5
* starting on patch * finish determining all required patches * properly redefine the patch rules * new patch * rem double semicolon * fix patch file * Merge branch 'master' of github.com:prysmaticlabs/prysm into deprecate-eth-protos * building the deps * test target passes using ethereumapis * compile gateway * attempting to build everything * e2e use ethereumapis * more fixes for slasher * other item * getting closer to compiling slasher * build slasher package * Merge branch 'master' into deprecate-eth-protos * Merge branch 'master' into deprecate-eth-protos * fix benches * lint gazelle * Merge branch 'deprecate-eth-protos' of github.com:prysmaticlabs/prysm into deprecate-eth-protos * proper gateway * lint * Merge branch 'master' into deprecate-eth-protos * fix build * Merge branch 'deprecate-eth-protos' of github.com:prysmaticlabs/prysm into deprecate-eth-protos * use swag * resolve * ignore change * include new patch changes * fix test * builds * fix e2e * gaz
267 lines
9.5 KiB
Go
267 lines
9.5 KiB
Go
// Package rpc defines the services that the beacon-chain uses to communicate via gRPC.
|
|
package rpc
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"math/rand"
|
|
"net"
|
|
"os"
|
|
|
|
middleware "github.com/grpc-ecosystem/go-grpc-middleware"
|
|
recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
|
|
grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
|
|
grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
|
|
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/cache"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/cache/depositcache"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/core/statefeed"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/db"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/operations"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/powchain"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/aggregator"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/attester"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/beacon"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/node"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/proposer"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/rpc/validator"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/sync"
|
|
pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
|
|
pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
|
"github.com/prysmaticlabs/prysm/shared/traceutil"
|
|
"github.com/sirupsen/logrus"
|
|
"go.opencensus.io/plugin/ocgrpc"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/reflection"
|
|
)
|
|
|
|
var log logrus.FieldLogger
|
|
|
|
func init() {
|
|
log = logrus.WithField("prefix", "rpc")
|
|
rand.Seed(int64(os.Getpid()))
|
|
}
|
|
|
|
// Service defining an RPC server for a beacon node.
|
|
type Service struct {
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
beaconDB db.Database
|
|
headFetcher blockchain.HeadFetcher
|
|
forkFetcher blockchain.ForkFetcher
|
|
finalizationFetcher blockchain.FinalizationFetcher
|
|
genesisTimeFetcher blockchain.GenesisTimeFetcher
|
|
attestationReceiver blockchain.AttestationReceiver
|
|
blockReceiver blockchain.BlockReceiver
|
|
powChainService powchain.Chain
|
|
chainStartFetcher powchain.ChainStartFetcher
|
|
mockEth1Votes bool
|
|
attestationsPool operations.Pool
|
|
operationsHandler operations.Handler
|
|
syncService sync.Checker
|
|
port string
|
|
listener net.Listener
|
|
withCert string
|
|
withKey string
|
|
grpcServer *grpc.Server
|
|
canonicalStateChan chan *pbp2p.BeaconState
|
|
incomingAttestation chan *ethpb.Attestation
|
|
credentialError error
|
|
p2p p2p.Broadcaster
|
|
depositFetcher depositcache.DepositFetcher
|
|
pendingDepositFetcher depositcache.PendingDepositsFetcher
|
|
stateNotifier statefeed.Notifier
|
|
}
|
|
|
|
// Config options for the beacon node RPC server.
|
|
type Config struct {
|
|
Port string
|
|
CertFlag string
|
|
KeyFlag string
|
|
BeaconDB db.Database
|
|
HeadFetcher blockchain.HeadFetcher
|
|
ForkFetcher blockchain.ForkFetcher
|
|
FinalizationFetcher blockchain.FinalizationFetcher
|
|
AttestationReceiver blockchain.AttestationReceiver
|
|
BlockReceiver blockchain.BlockReceiver
|
|
POWChainService powchain.Chain
|
|
ChainStartFetcher powchain.ChainStartFetcher
|
|
GenesisTimeFetcher blockchain.GenesisTimeFetcher
|
|
MockEth1Votes bool
|
|
OperationsHandler operations.Handler
|
|
AttestationsPool operations.Pool
|
|
SyncService sync.Checker
|
|
Broadcaster p2p.Broadcaster
|
|
DepositFetcher depositcache.DepositFetcher
|
|
PendingDepositFetcher depositcache.PendingDepositsFetcher
|
|
StateNotifier statefeed.Notifier
|
|
}
|
|
|
|
// NewService instantiates a new RPC service instance that will
|
|
// be registered into a running beacon node.
|
|
func NewService(ctx context.Context, cfg *Config) *Service {
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
return &Service{
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
beaconDB: cfg.BeaconDB,
|
|
headFetcher: cfg.HeadFetcher,
|
|
forkFetcher: cfg.ForkFetcher,
|
|
finalizationFetcher: cfg.FinalizationFetcher,
|
|
genesisTimeFetcher: cfg.GenesisTimeFetcher,
|
|
attestationReceiver: cfg.AttestationReceiver,
|
|
blockReceiver: cfg.BlockReceiver,
|
|
p2p: cfg.Broadcaster,
|
|
powChainService: cfg.POWChainService,
|
|
chainStartFetcher: cfg.ChainStartFetcher,
|
|
mockEth1Votes: cfg.MockEth1Votes,
|
|
attestationsPool: cfg.AttestationsPool,
|
|
operationsHandler: cfg.OperationsHandler,
|
|
syncService: cfg.SyncService,
|
|
port: cfg.Port,
|
|
withCert: cfg.CertFlag,
|
|
withKey: cfg.KeyFlag,
|
|
depositFetcher: cfg.DepositFetcher,
|
|
pendingDepositFetcher: cfg.PendingDepositFetcher,
|
|
canonicalStateChan: make(chan *pbp2p.BeaconState, params.BeaconConfig().DefaultBufferSize),
|
|
incomingAttestation: make(chan *ethpb.Attestation, params.BeaconConfig().DefaultBufferSize),
|
|
stateNotifier: cfg.StateNotifier,
|
|
}
|
|
}
|
|
|
|
// Start the gRPC server.
|
|
func (s *Service) Start() {
|
|
lis, err := net.Listen("tcp", fmt.Sprintf(":%s", s.port))
|
|
if err != nil {
|
|
log.Errorf("Could not listen to port in Start() :%s: %v", s.port, err)
|
|
}
|
|
s.listener = lis
|
|
log.WithField("port", fmt.Sprintf(":%s", s.port)).Info("RPC-API listening on port")
|
|
|
|
opts := []grpc.ServerOption{
|
|
grpc.StatsHandler(&ocgrpc.ServerHandler{}),
|
|
grpc.StreamInterceptor(middleware.ChainStreamServer(
|
|
recovery.StreamServerInterceptor(
|
|
recovery.WithRecoveryHandlerContext(traceutil.RecoveryHandlerFunc),
|
|
),
|
|
grpc_prometheus.StreamServerInterceptor,
|
|
grpc_opentracing.StreamServerInterceptor(),
|
|
)),
|
|
grpc.UnaryInterceptor(middleware.ChainUnaryServer(
|
|
recovery.UnaryServerInterceptor(
|
|
recovery.WithRecoveryHandlerContext(traceutil.RecoveryHandlerFunc),
|
|
),
|
|
grpc_prometheus.UnaryServerInterceptor,
|
|
grpc_opentracing.UnaryServerInterceptor(),
|
|
)),
|
|
}
|
|
// TODO(#791): Utilize a certificate for secure connections
|
|
// between beacon nodes and validator clients.
|
|
if s.withCert != "" && s.withKey != "" {
|
|
creds, err := credentials.NewServerTLSFromFile(s.withCert, s.withKey)
|
|
if err != nil {
|
|
log.Errorf("Could not load TLS keys: %s", err)
|
|
s.credentialError = err
|
|
}
|
|
opts = append(opts, grpc.Creds(creds))
|
|
} else {
|
|
log.Warn("You are using an insecure gRPC connection! Provide a certificate and key to connect securely")
|
|
}
|
|
s.grpcServer = grpc.NewServer(opts...)
|
|
|
|
proposerServer := &proposer.Server{
|
|
BeaconDB: s.beaconDB,
|
|
HeadFetcher: s.headFetcher,
|
|
BlockReceiver: s.blockReceiver,
|
|
ChainStartFetcher: s.chainStartFetcher,
|
|
Eth1InfoFetcher: s.powChainService,
|
|
Eth1BlockFetcher: s.powChainService,
|
|
MockEth1Votes: s.mockEth1Votes,
|
|
Pool: s.attestationsPool,
|
|
CanonicalStateChan: s.canonicalStateChan,
|
|
DepositFetcher: s.depositFetcher,
|
|
PendingDepositsFetcher: s.pendingDepositFetcher,
|
|
SyncChecker: s.syncService,
|
|
}
|
|
attesterServer := &attester.Server{
|
|
P2p: s.p2p,
|
|
BeaconDB: s.beaconDB,
|
|
OperationsHandler: s.operationsHandler,
|
|
AttReceiver: s.attestationReceiver,
|
|
HeadFetcher: s.headFetcher,
|
|
AttestationCache: cache.NewAttestationCache(),
|
|
SyncChecker: s.syncService,
|
|
}
|
|
validatorServer := &validator.Server{
|
|
Ctx: s.ctx,
|
|
BeaconDB: s.beaconDB,
|
|
HeadFetcher: s.headFetcher,
|
|
ForkFetcher: s.forkFetcher,
|
|
CanonicalStateChan: s.canonicalStateChan,
|
|
BlockFetcher: s.powChainService,
|
|
ChainStartFetcher: s.chainStartFetcher,
|
|
Eth1InfoFetcher: s.powChainService,
|
|
DepositFetcher: s.depositFetcher,
|
|
SyncChecker: s.syncService,
|
|
StateNotifier: s.stateNotifier,
|
|
}
|
|
nodeServer := &node.Server{
|
|
BeaconDB: s.beaconDB,
|
|
Server: s.grpcServer,
|
|
SyncChecker: s.syncService,
|
|
GenesisTimeFetcher: s.genesisTimeFetcher,
|
|
}
|
|
beaconChainServer := &beacon.Server{
|
|
BeaconDB: s.beaconDB,
|
|
Pool: s.attestationsPool,
|
|
HeadFetcher: s.headFetcher,
|
|
FinalizationFetcher: s.finalizationFetcher,
|
|
ChainStartFetcher: s.chainStartFetcher,
|
|
CanonicalStateChan: s.canonicalStateChan,
|
|
}
|
|
aggregatorServer := &aggregator.Server{
|
|
BeaconDB: s.beaconDB,
|
|
HeadFetcher: s.headFetcher,
|
|
SyncChecker: s.syncService,
|
|
}
|
|
pb.RegisterProposerServiceServer(s.grpcServer, proposerServer)
|
|
pb.RegisterAttesterServiceServer(s.grpcServer, attesterServer)
|
|
pb.RegisterValidatorServiceServer(s.grpcServer, validatorServer)
|
|
pb.RegisterAggregatorServiceServer(s.grpcServer, aggregatorServer)
|
|
ethpb.RegisterNodeServer(s.grpcServer, nodeServer)
|
|
ethpb.RegisterBeaconChainServer(s.grpcServer, beaconChainServer)
|
|
|
|
// Register reflection service on gRPC server.
|
|
reflection.Register(s.grpcServer)
|
|
|
|
go func() {
|
|
if s.listener != nil {
|
|
if err := s.grpcServer.Serve(s.listener); err != nil {
|
|
log.Errorf("Could not serve gRPC: %v", err)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Stop the service.
|
|
func (s *Service) Stop() error {
|
|
s.cancel()
|
|
if s.listener != nil {
|
|
s.grpcServer.GracefulStop()
|
|
log.Debug("Initiated graceful stop of gRPC server")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Status returns nil or credentialError
|
|
func (s *Service) Status() error {
|
|
if s.credentialError != nil {
|
|
return s.credentialError
|
|
}
|
|
return nil
|
|
}
|