/
server.go
306 lines (287 loc) · 10 KB
/
server.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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
// Package node defines a gRPC node service implementation, providing
// useful endpoints for checking a node's sync status, peer info,
// genesis data, and version information.
package node
import (
"context"
"fmt"
"net/http"
"sort"
"strconv"
"time"
"github.com/golang/protobuf/ptypes/empty"
"github.com/golang/protobuf/ptypes/timestamp"
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/blockchain"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/db"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/execution"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/p2p"
"github.com/prysmaticlabs/prysm/v5/beacon-chain/sync"
"github.com/prysmaticlabs/prysm/v5/io/logs"
ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
"github.com/prysmaticlabs/prysm/v5/runtime/version"
"go.opencensus.io/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/timestamppb"
)
// Server defines a server implementation of the gRPC Node service,
// providing RPC endpoints for verifying a beacon node's sync status, genesis and
// version information, and services the node implements and runs.
type Server struct {
LogsStreamer logs.Streamer
StreamLogsBufferSize int
SyncChecker sync.Checker
Server *grpc.Server
BeaconDB db.ReadOnlyDatabase
PeersFetcher p2p.PeersProvider
PeerManager p2p.PeerManager
GenesisTimeFetcher blockchain.TimeFetcher
GenesisFetcher blockchain.GenesisFetcher
POWChainInfoFetcher execution.ChainInfoFetcher
BeaconMonitoringHost string
BeaconMonitoringPort int
}
// GetHealth checks the health of the node
func (ns *Server) GetHealth(ctx context.Context, request *ethpb.HealthRequest) (*empty.Empty, error) {
ctx, span := trace.StartSpan(ctx, "node.GetHealth")
defer span.End()
// Set a timeout for the health check operation
timeoutDuration := 10 * time.Second
ctx, cancel := context.WithTimeout(ctx, timeoutDuration)
defer cancel() // Important to avoid a context leak
if ns.SyncChecker.Synced() {
return &empty.Empty{}, nil
}
if ns.SyncChecker.Syncing() || ns.SyncChecker.Initialized() {
if request.SyncingStatus != 0 {
// override the 200 success with the provided request status
if err := grpc.SetHeader(ctx, metadata.Pairs("x-http-code", strconv.FormatUint(request.SyncingStatus, 10))); err != nil {
return &empty.Empty{}, status.Errorf(codes.Internal, "Could not set custom success code header: %v", err)
}
return &empty.Empty{}, nil
}
if err := grpc.SetHeader(ctx, metadata.Pairs("x-http-code", strconv.FormatUint(http.StatusPartialContent, 10))); err != nil {
return &empty.Empty{}, status.Errorf(codes.Internal, "Could not set custom success code header: %v", err)
}
return &empty.Empty{}, nil
}
return &empty.Empty{}, status.Errorf(codes.Unavailable, "service unavailable")
}
// GetSyncStatus checks the current network sync status of the node.
func (ns *Server) GetSyncStatus(_ context.Context, _ *empty.Empty) (*ethpb.SyncStatus, error) {
return ðpb.SyncStatus{
Syncing: ns.SyncChecker.Syncing(),
}, nil
}
// GetGenesis fetches genesis chain information of Ethereum. Returns unix timestamp 0
// if a genesis time has yet to be determined.
func (ns *Server) GetGenesis(ctx context.Context, _ *empty.Empty) (*ethpb.Genesis, error) {
contractAddr, err := ns.BeaconDB.DepositContractAddress(ctx)
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not retrieve contract address from db: %v", err)
}
genesisTime := ns.GenesisTimeFetcher.GenesisTime()
var defaultGenesisTime time.Time
var gt *timestamp.Timestamp
if genesisTime == defaultGenesisTime {
gt = timestamppb.New(time.Unix(0, 0))
} else {
gt = timestamppb.New(genesisTime)
}
genValRoot := ns.GenesisFetcher.GenesisValidatorsRoot()
return ðpb.Genesis{
GenesisTime: gt,
DepositContractAddress: contractAddr,
GenesisValidatorsRoot: genValRoot[:],
}, nil
}
// GetVersion checks the version information of the beacon node.
func (_ *Server) GetVersion(_ context.Context, _ *empty.Empty) (*ethpb.Version, error) {
return ðpb.Version{
Version: version.Version(),
}, nil
}
// ListImplementedServices lists the services implemented and enabled by this node.
//
// Any service not present in this list may return UNIMPLEMENTED or
// PERMISSION_DENIED. The server may also support fetching services by grpc
// reflection.
func (ns *Server) ListImplementedServices(_ context.Context, _ *empty.Empty) (*ethpb.ImplementedServices, error) {
serviceInfo := ns.Server.GetServiceInfo()
serviceNames := make([]string, 0, len(serviceInfo))
for svc := range serviceInfo {
serviceNames = append(serviceNames, svc)
}
sort.Strings(serviceNames)
return ðpb.ImplementedServices{
Services: serviceNames,
}, nil
}
// GetHost returns the p2p data on the current local and host peer.
func (ns *Server) GetHost(_ context.Context, _ *empty.Empty) (*ethpb.HostData, error) {
var stringAddr []string
for _, addr := range ns.PeerManager.Host().Addrs() {
stringAddr = append(stringAddr, addr.String())
}
record := ns.PeerManager.ENR()
enr := ""
var err error
if record != nil {
enr, err = p2p.SerializeENR(record)
if err != nil {
return nil, status.Errorf(codes.Internal, "Unable to serialize enr: %v", err)
}
}
return ðpb.HostData{
Addresses: stringAddr,
PeerId: ns.PeerManager.PeerID().String(),
Enr: enr,
}, nil
}
// GetPeer returns the data known about the peer defined by the provided peer id.
func (ns *Server) GetPeer(_ context.Context, peerReq *ethpb.PeerRequest) (*ethpb.Peer, error) {
pid, err := peer.Decode(peerReq.PeerId)
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "Unable to parse provided peer id: %v", err)
}
addr, err := ns.PeersFetcher.Peers().Address(pid)
if err != nil {
return nil, status.Errorf(codes.NotFound, "Requested peer does not exist: %v", err)
}
dir, err := ns.PeersFetcher.Peers().Direction(pid)
if err != nil {
return nil, status.Errorf(codes.NotFound, "Requested peer does not exist: %v", err)
}
pbDirection := ethpb.PeerDirection_UNKNOWN
switch dir {
case network.DirInbound:
pbDirection = ethpb.PeerDirection_INBOUND
case network.DirOutbound:
pbDirection = ethpb.PeerDirection_OUTBOUND
}
connState, err := ns.PeersFetcher.Peers().ConnectionState(pid)
if err != nil {
return nil, status.Errorf(codes.NotFound, "Requested peer does not exist: %v", err)
}
record, err := ns.PeersFetcher.Peers().ENR(pid)
if err != nil {
return nil, status.Errorf(codes.NotFound, "Requested peer does not exist: %v", err)
}
enr := ""
if record != nil {
enr, err = p2p.SerializeENR(record)
if err != nil {
return nil, status.Errorf(codes.Internal, "Unable to serialize enr: %v", err)
}
}
return ðpb.Peer{
Address: addr.String(),
Direction: pbDirection,
ConnectionState: ethpb.ConnectionState(connState),
PeerId: peerReq.PeerId,
Enr: enr,
}, nil
}
// ListPeers lists the peers connected to this node.
func (ns *Server) ListPeers(ctx context.Context, _ *empty.Empty) (*ethpb.Peers, error) {
peers := ns.PeersFetcher.Peers().Connected()
res := make([]*ethpb.Peer, 0, len(peers))
for _, pid := range peers {
if ctx.Err() != nil {
return nil, ctx.Err()
}
multiaddr, err := ns.PeersFetcher.Peers().Address(pid)
if err != nil {
continue
}
direction, err := ns.PeersFetcher.Peers().Direction(pid)
if err != nil {
continue
}
record, err := ns.PeersFetcher.Peers().ENR(pid)
if err != nil {
continue
}
enr := ""
if record != nil {
enr, err = p2p.SerializeENR(record)
if err != nil {
continue
}
}
multiAddrStr := "unknown"
if multiaddr != nil {
multiAddrStr = multiaddr.String()
}
address := fmt.Sprintf("%s/p2p/%s", multiAddrStr, pid.String())
pbDirection := ethpb.PeerDirection_UNKNOWN
switch direction {
case network.DirInbound:
pbDirection = ethpb.PeerDirection_INBOUND
case network.DirOutbound:
pbDirection = ethpb.PeerDirection_OUTBOUND
}
res = append(res, ðpb.Peer{
Address: address,
Direction: pbDirection,
ConnectionState: ethpb.ConnectionState_CONNECTED,
PeerId: pid.String(),
Enr: enr,
})
}
return ðpb.Peers{
Peers: res,
}, nil
}
// GetETH1ConnectionStatus gets data about the ETH1 endpoints.
func (ns *Server) GetETH1ConnectionStatus(_ context.Context, _ *empty.Empty) (*ethpb.ETH1ConnectionStatus, error) {
var currErr string
err := ns.POWChainInfoFetcher.ExecutionClientConnectionErr()
if err != nil {
currErr = err.Error()
}
return ðpb.ETH1ConnectionStatus{
CurrentAddress: ns.POWChainInfoFetcher.ExecutionClientEndpoint(),
CurrentConnectionError: currErr,
Addresses: []string{ns.POWChainInfoFetcher.ExecutionClientEndpoint()},
}, nil
}
// StreamBeaconLogs from the beacon node via a gRPC server-side stream.
// DEPRECATED: This endpoint doesn't appear to be used and have been marked for deprecation.
func (ns *Server) StreamBeaconLogs(_ *empty.Empty, stream ethpb.Health_StreamBeaconLogsServer) error {
ch := make(chan []byte, ns.StreamLogsBufferSize)
sub := ns.LogsStreamer.LogsFeed().Subscribe(ch)
defer func() {
sub.Unsubscribe()
close(ch)
}()
recentLogs := ns.LogsStreamer.GetLastFewLogs()
logStrings := make([]string, len(recentLogs))
for i, log := range recentLogs {
logStrings[i] = string(log)
}
if err := stream.Send(ðpb.LogsResponse{
Logs: logStrings,
}); err != nil {
return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
}
for {
select {
case log := <-ch:
resp := ðpb.LogsResponse{
Logs: []string{string(log)},
}
if err := stream.Send(resp); err != nil {
return status.Errorf(codes.Unavailable, "Could not send over stream: %v", err)
}
case err := <-sub.Err():
return status.Errorf(codes.Canceled, "Subscriber error, closing: %v", err)
case <-stream.Context().Done():
return status.Error(codes.Canceled, "Context canceled")
}
}
}