forked from prysmaticlabs/prysm
-
Notifications
You must be signed in to change notification settings - Fork 0
/
service.go
180 lines (163 loc) · 5.32 KB
/
service.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
// Package rpc defines the services that the beacon-chain uses to communicate via gRPC.
package rpc
import (
"context"
"fmt"
"math/big"
"net"
"time"
"github.com/prysmaticlabs/prysm/beacon-chain/db"
pbp2p "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
pb "github.com/prysmaticlabs/prysm/proto/beacon/rpc/v1"
"github.com/prysmaticlabs/prysm/shared/event"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/sirupsen/logrus"
"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")
}
type chainService interface {
IncomingBlockFeed() *event.Feed
// These methods are not called on-demand by a validator
// but instead streamed to connected validators every
// time the canonical head changes in the chain service.
CanonicalBlockFeed() *event.Feed
CanonicalStateFeed() *event.Feed
}
type operationService interface {
IncomingExitFeed() *event.Feed
IncomingAttFeed() *event.Feed
}
type powChainService interface {
HasChainStartLogOccurred() (bool, uint64, error)
ChainStartFeed() *event.Feed
LatestBlockNumber() *big.Int
}
// Service defining an RPC server for a beacon node.
type Service struct {
ctx context.Context
cancel context.CancelFunc
beaconDB *db.BeaconDB
chainService chainService
powChainService powChainService
operationService operationService
port string
listener net.Listener
withCert string
withKey string
grpcServer *grpc.Server
canonicalBlockChan chan *pbp2p.BeaconBlock
canonicalStateChan chan *pbp2p.BeaconState
incomingAttestation chan *pbp2p.Attestation
slotAlignmentDuration time.Duration
credentialError error
}
// Config options for the beacon node RPC server.
type Config struct {
Port string
CertFlag string
KeyFlag string
SubscriptionBuf int
BeaconDB *db.BeaconDB
ChainService chainService
POWChainService powChainService
OperationService operationService
}
// NewRPCService creates a new instance of a struct implementing the BeaconServiceServer
// interface.
func NewRPCService(ctx context.Context, cfg *Config) *Service {
ctx, cancel := context.WithCancel(ctx)
return &Service{
ctx: ctx,
cancel: cancel,
beaconDB: cfg.BeaconDB,
chainService: cfg.ChainService,
powChainService: cfg.POWChainService,
operationService: cfg.OperationService,
port: cfg.Port,
withCert: cfg.CertFlag,
withKey: cfg.KeyFlag,
slotAlignmentDuration: time.Duration(params.BeaconConfig().SlotDuration) * time.Second,
canonicalBlockChan: make(chan *pbp2p.BeaconBlock, cfg.SubscriptionBuf),
canonicalStateChan: make(chan *pbp2p.BeaconState, cfg.SubscriptionBuf),
incomingAttestation: make(chan *pbp2p.Attestation, cfg.SubscriptionBuf),
}
}
// Start the gRPC server.
func (s *Service) Start() {
log.Info("Starting service")
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.Infof("RPC server listening on port :%s", s.port)
// 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
}
s.grpcServer = grpc.NewServer(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()
}
beaconServer := &BeaconServer{
beaconDB: s.beaconDB,
ctx: s.ctx,
powChainService: s.powChainService,
operationService: s.operationService,
incomingAttestation: s.incomingAttestation,
canonicalStateChan: s.canonicalStateChan,
chainStartChan: make(chan time.Time, 1),
}
proposerServer := &ProposerServer{
beaconDB: s.beaconDB,
chainService: s.chainService,
powChainService: s.powChainService,
canonicalStateChan: s.canonicalStateChan,
}
attesterServer := &AttesterServer{
operationService: s.operationService,
}
validatorServer := &ValidatorServer{
beaconDB: s.beaconDB,
}
pb.RegisterBeaconServiceServer(s.grpcServer, beaconServer)
pb.RegisterProposerServiceServer(s.grpcServer, proposerServer)
pb.RegisterAttesterServiceServer(s.grpcServer, attesterServer)
pb.RegisterValidatorServiceServer(s.grpcServer, validatorServer)
// 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 {
log.Info("Stopping service")
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
}