ethstats.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811
  1. // Copyright 2016 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. // Package ethstats implements the network stats reporting service.
  17. package ethstats
  18. import (
  19. "context"
  20. "encoding/json"
  21. "errors"
  22. "fmt"
  23. "math/big"
  24. "net/http"
  25. "runtime"
  26. "strconv"
  27. "strings"
  28. "sync"
  29. "time"
  30. "github.com/ethereum/go-ethereum"
  31. "github.com/ethereum/go-ethereum/common"
  32. "github.com/ethereum/go-ethereum/common/mclock"
  33. "github.com/ethereum/go-ethereum/consensus"
  34. "github.com/ethereum/go-ethereum/core"
  35. "github.com/ethereum/go-ethereum/core/types"
  36. ethproto "github.com/ethereum/go-ethereum/eth/protocols/eth"
  37. "github.com/ethereum/go-ethereum/event"
  38. "github.com/ethereum/go-ethereum/les"
  39. "github.com/ethereum/go-ethereum/log"
  40. "github.com/ethereum/go-ethereum/miner"
  41. "github.com/ethereum/go-ethereum/node"
  42. "github.com/ethereum/go-ethereum/p2p"
  43. "github.com/ethereum/go-ethereum/rpc"
  44. "github.com/gorilla/websocket"
  45. )
  46. const (
  47. // historyUpdateRange is the number of blocks a node should report upon login or
  48. // history request.
  49. historyUpdateRange = 50
  50. // txChanSize is the size of channel listening to NewTxsEvent.
  51. // The number is referenced from the size of tx pool.
  52. txChanSize = 4096
  53. // chainHeadChanSize is the size of channel listening to ChainHeadEvent.
  54. chainHeadChanSize = 10
  55. )
  56. // backend encompasses the bare-minimum functionality needed for ethstats reporting
  57. type backend interface {
  58. SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
  59. SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription
  60. CurrentHeader() *types.Header
  61. HeaderByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Header, error)
  62. GetTd(ctx context.Context, hash common.Hash) *big.Int
  63. Stats() (pending int, queued int)
  64. SyncProgress() ethereum.SyncProgress
  65. }
  66. // fullNodeBackend encompasses the functionality necessary for a full node
  67. // reporting to ethstats
  68. type fullNodeBackend interface {
  69. backend
  70. Miner() *miner.Miner
  71. BlockByNumber(ctx context.Context, number rpc.BlockNumber) (*types.Block, error)
  72. CurrentBlock() *types.Block
  73. SuggestGasTipCap(ctx context.Context) (*big.Int, error)
  74. }
  75. // Service implements an Ethereum netstats reporting daemon that pushes local
  76. // chain statistics up to a monitoring server.
  77. type Service struct {
  78. server *p2p.Server // Peer-to-peer server to retrieve networking infos
  79. backend backend
  80. engine consensus.Engine // Consensus engine to retrieve variadic block fields
  81. node string // Name of the node to display on the monitoring page
  82. pass string // Password to authorize access to the monitoring page
  83. host string // Remote address of the monitoring service
  84. pongCh chan struct{} // Pong notifications are fed into this channel
  85. histCh chan []uint64 // History request block numbers are fed into this channel
  86. headSub event.Subscription
  87. txSub event.Subscription
  88. }
  89. // connWrapper is a wrapper to prevent concurrent-write or concurrent-read on the
  90. // websocket.
  91. //
  92. // From Gorilla websocket docs:
  93. // Connections support one concurrent reader and one concurrent writer.
  94. // Applications are responsible for ensuring that no more than one goroutine calls the write methods
  95. // - NextWriter, SetWriteDeadline, WriteMessage, WriteJSON, EnableWriteCompression, SetCompressionLevel
  96. // concurrently and that no more than one goroutine calls the read methods
  97. // - NextReader, SetReadDeadline, ReadMessage, ReadJSON, SetPongHandler, SetPingHandler
  98. // concurrently.
  99. // The Close and WriteControl methods can be called concurrently with all other methods.
  100. type connWrapper struct {
  101. conn *websocket.Conn
  102. rlock sync.Mutex
  103. wlock sync.Mutex
  104. }
  105. func newConnectionWrapper(conn *websocket.Conn) *connWrapper {
  106. return &connWrapper{conn: conn}
  107. }
  108. // WriteJSON wraps corresponding method on the websocket but is safe for concurrent calling
  109. func (w *connWrapper) WriteJSON(v interface{}) error {
  110. w.wlock.Lock()
  111. defer w.wlock.Unlock()
  112. return w.conn.WriteJSON(v)
  113. }
  114. // ReadJSON wraps corresponding method on the websocket but is safe for concurrent calling
  115. func (w *connWrapper) ReadJSON(v interface{}) error {
  116. w.rlock.Lock()
  117. defer w.rlock.Unlock()
  118. return w.conn.ReadJSON(v)
  119. }
  120. // Close wraps corresponding method on the websocket but is safe for concurrent calling
  121. func (w *connWrapper) Close() error {
  122. // The Close and WriteControl methods can be called concurrently with all other methods,
  123. // so the mutex is not used here
  124. return w.conn.Close()
  125. }
  126. // parseEthstatsURL parses the netstats connection url.
  127. // URL argument should be of the form <nodename:secret@host:port>
  128. // If non-erroring, the returned slice contains 3 elements: [nodename, pass, host]
  129. func parseEthstatsURL(url string) (parts []string, err error) {
  130. err = fmt.Errorf("invalid netstats url: \"%s\", should be nodename:secret@host:port", url)
  131. hostIndex := strings.LastIndex(url, "@")
  132. if hostIndex == -1 || hostIndex == len(url)-1 {
  133. return nil, err
  134. }
  135. preHost, host := url[:hostIndex], url[hostIndex+1:]
  136. passIndex := strings.LastIndex(preHost, ":")
  137. if passIndex == -1 {
  138. return []string{preHost, "", host}, nil
  139. }
  140. nodename, pass := preHost[:passIndex], ""
  141. if passIndex != len(preHost)-1 {
  142. pass = preHost[passIndex+1:]
  143. }
  144. return []string{nodename, pass, host}, nil
  145. }
  146. // New returns a monitoring service ready for stats reporting.
  147. func New(node *node.Node, backend backend, engine consensus.Engine, url string) error {
  148. parts, err := parseEthstatsURL(url)
  149. if err != nil {
  150. return err
  151. }
  152. ethstats := &Service{
  153. backend: backend,
  154. engine: engine,
  155. server: node.Server(),
  156. node: parts[0],
  157. pass: parts[1],
  158. host: parts[2],
  159. pongCh: make(chan struct{}),
  160. histCh: make(chan []uint64, 1),
  161. }
  162. node.RegisterLifecycle(ethstats)
  163. return nil
  164. }
  165. // Start implements node.Lifecycle, starting up the monitoring and reporting daemon.
  166. func (s *Service) Start() error {
  167. // Subscribe to chain events to execute updates on
  168. chainHeadCh := make(chan core.ChainHeadEvent, chainHeadChanSize)
  169. s.headSub = s.backend.SubscribeChainHeadEvent(chainHeadCh)
  170. txEventCh := make(chan core.NewTxsEvent, txChanSize)
  171. s.txSub = s.backend.SubscribeNewTxsEvent(txEventCh)
  172. go s.loop(chainHeadCh, txEventCh)
  173. log.Info("Stats daemon started")
  174. return nil
  175. }
  176. // Stop implements node.Lifecycle, terminating the monitoring and reporting daemon.
  177. func (s *Service) Stop() error {
  178. s.headSub.Unsubscribe()
  179. s.txSub.Unsubscribe()
  180. log.Info("Stats daemon stopped")
  181. return nil
  182. }
  183. // loop keeps trying to connect to the netstats server, reporting chain events
  184. // until termination.
  185. func (s *Service) loop(chainHeadCh chan core.ChainHeadEvent, txEventCh chan core.NewTxsEvent) {
  186. // Start a goroutine that exhausts the subscriptions to avoid events piling up
  187. var (
  188. quitCh = make(chan struct{})
  189. headCh = make(chan *types.Block, 1)
  190. txCh = make(chan struct{}, 1)
  191. )
  192. go func() {
  193. var lastTx mclock.AbsTime
  194. HandleLoop:
  195. for {
  196. select {
  197. // Notify of chain head events, but drop if too frequent
  198. case head := <-chainHeadCh:
  199. select {
  200. case headCh <- head.Block:
  201. default:
  202. }
  203. // Notify of new transaction events, but drop if too frequent
  204. case <-txEventCh:
  205. if time.Duration(mclock.Now()-lastTx) < time.Second {
  206. continue
  207. }
  208. lastTx = mclock.Now()
  209. select {
  210. case txCh <- struct{}{}:
  211. default:
  212. }
  213. // node stopped
  214. case <-s.txSub.Err():
  215. break HandleLoop
  216. case <-s.headSub.Err():
  217. break HandleLoop
  218. }
  219. }
  220. close(quitCh)
  221. }()
  222. // Resolve the URL, defaulting to TLS, but falling back to none too
  223. path := fmt.Sprintf("%s/api", s.host)
  224. urls := []string{path}
  225. // url.Parse and url.IsAbs is unsuitable (https://github.com/golang/go/issues/19779)
  226. if !strings.Contains(path, "://") {
  227. urls = []string{"wss://" + path, "ws://" + path}
  228. }
  229. errTimer := time.NewTimer(0)
  230. defer errTimer.Stop()
  231. // Loop reporting until termination
  232. for {
  233. select {
  234. case <-quitCh:
  235. return
  236. case <-errTimer.C:
  237. // Establish a websocket connection to the server on any supported URL
  238. var (
  239. conn *connWrapper
  240. err error
  241. )
  242. dialer := websocket.Dialer{HandshakeTimeout: 5 * time.Second}
  243. header := make(http.Header)
  244. header.Set("origin", "http://localhost")
  245. for _, url := range urls {
  246. c, _, e := dialer.Dial(url, header)
  247. err = e
  248. if err == nil {
  249. conn = newConnectionWrapper(c)
  250. break
  251. }
  252. }
  253. if err != nil {
  254. log.Warn("Stats server unreachable", "err", err)
  255. errTimer.Reset(10 * time.Second)
  256. continue
  257. }
  258. // Authenticate the client with the server
  259. if err = s.login(conn); err != nil {
  260. log.Warn("Stats login failed", "err", err)
  261. conn.Close()
  262. errTimer.Reset(10 * time.Second)
  263. continue
  264. }
  265. go s.readLoop(conn)
  266. // Send the initial stats so our node looks decent from the get go
  267. if err = s.report(conn); err != nil {
  268. log.Warn("Initial stats report failed", "err", err)
  269. conn.Close()
  270. errTimer.Reset(0)
  271. continue
  272. }
  273. // Keep sending status updates until the connection breaks
  274. fullReport := time.NewTicker(15 * time.Second)
  275. for err == nil {
  276. select {
  277. case <-quitCh:
  278. fullReport.Stop()
  279. // Make sure the connection is closed
  280. conn.Close()
  281. return
  282. case <-fullReport.C:
  283. if err = s.report(conn); err != nil {
  284. log.Warn("Full stats report failed", "err", err)
  285. }
  286. case list := <-s.histCh:
  287. if err = s.reportHistory(conn, list); err != nil {
  288. log.Warn("Requested history report failed", "err", err)
  289. }
  290. case head := <-headCh:
  291. if err = s.reportBlock(conn, head); err != nil {
  292. log.Warn("Block stats report failed", "err", err)
  293. }
  294. if err = s.reportPending(conn); err != nil {
  295. log.Warn("Post-block transaction stats report failed", "err", err)
  296. }
  297. case <-txCh:
  298. if err = s.reportPending(conn); err != nil {
  299. log.Warn("Transaction stats report failed", "err", err)
  300. }
  301. }
  302. }
  303. fullReport.Stop()
  304. // Close the current connection and establish a new one
  305. conn.Close()
  306. errTimer.Reset(0)
  307. }
  308. }
  309. }
  310. // readLoop loops as long as the connection is alive and retrieves data packets
  311. // from the network socket. If any of them match an active request, it forwards
  312. // it, if they themselves are requests it initiates a reply, and lastly it drops
  313. // unknown packets.
  314. func (s *Service) readLoop(conn *connWrapper) {
  315. // If the read loop exits, close the connection
  316. defer conn.Close()
  317. for {
  318. // Retrieve the next generic network packet and bail out on error
  319. var blob json.RawMessage
  320. if err := conn.ReadJSON(&blob); err != nil {
  321. log.Warn("Failed to retrieve stats server message", "err", err)
  322. return
  323. }
  324. // If the network packet is a system ping, respond to it directly
  325. var ping string
  326. if err := json.Unmarshal(blob, &ping); err == nil && strings.HasPrefix(ping, "primus::ping::") {
  327. if err := conn.WriteJSON(strings.ReplaceAll(ping, "ping", "pong")); err != nil {
  328. log.Warn("Failed to respond to system ping message", "err", err)
  329. return
  330. }
  331. continue
  332. }
  333. // Not a system ping, try to decode an actual state message
  334. var msg map[string][]interface{}
  335. if err := json.Unmarshal(blob, &msg); err != nil {
  336. log.Warn("Failed to decode stats server message", "err", err)
  337. return
  338. }
  339. log.Trace("Received message from stats server", "msg", msg)
  340. if len(msg["emit"]) == 0 {
  341. log.Warn("Stats server sent non-broadcast", "msg", msg)
  342. return
  343. }
  344. command, ok := msg["emit"][0].(string)
  345. if !ok {
  346. log.Warn("Invalid stats server message type", "type", msg["emit"][0])
  347. return
  348. }
  349. // If the message is a ping reply, deliver (someone must be listening!)
  350. if len(msg["emit"]) == 2 && command == "node-pong" {
  351. select {
  352. case s.pongCh <- struct{}{}:
  353. // Pong delivered, continue listening
  354. continue
  355. default:
  356. // Ping routine dead, abort
  357. log.Warn("Stats server pinger seems to have died")
  358. return
  359. }
  360. }
  361. // If the message is a history request, forward to the event processor
  362. if len(msg["emit"]) == 2 && command == "history" {
  363. // Make sure the request is valid and doesn't crash us
  364. request, ok := msg["emit"][1].(map[string]interface{})
  365. if !ok {
  366. log.Warn("Invalid stats history request", "msg", msg["emit"][1])
  367. select {
  368. case s.histCh <- nil: // Treat it as an no indexes request
  369. default:
  370. }
  371. continue
  372. }
  373. list, ok := request["list"].([]interface{})
  374. if !ok {
  375. log.Warn("Invalid stats history block list", "list", request["list"])
  376. return
  377. }
  378. // Convert the block number list to an integer list
  379. numbers := make([]uint64, len(list))
  380. for i, num := range list {
  381. n, ok := num.(float64)
  382. if !ok {
  383. log.Warn("Invalid stats history block number", "number", num)
  384. return
  385. }
  386. numbers[i] = uint64(n)
  387. }
  388. select {
  389. case s.histCh <- numbers:
  390. continue
  391. default:
  392. }
  393. }
  394. // Report anything else and continue
  395. log.Info("Unknown stats message", "msg", msg)
  396. }
  397. }
  398. // nodeInfo is the collection of meta information about a node that is displayed
  399. // on the monitoring page.
  400. type nodeInfo struct {
  401. Name string `json:"name"`
  402. Node string `json:"node"`
  403. Port int `json:"port"`
  404. Network string `json:"net"`
  405. Protocol string `json:"protocol"`
  406. API string `json:"api"`
  407. Os string `json:"os"`
  408. OsVer string `json:"os_v"`
  409. Client string `json:"client"`
  410. History bool `json:"canUpdateHistory"`
  411. }
  412. // authMsg is the authentication infos needed to login to a monitoring server.
  413. type authMsg struct {
  414. ID string `json:"id"`
  415. Info nodeInfo `json:"info"`
  416. Secret string `json:"secret"`
  417. }
  418. // login tries to authorize the client at the remote server.
  419. func (s *Service) login(conn *connWrapper) error {
  420. // Construct and send the login authentication
  421. infos := s.server.NodeInfo()
  422. var protocols []string
  423. for _, proto := range s.server.Protocols {
  424. protocols = append(protocols, fmt.Sprintf("%s/%d", proto.Name, proto.Version))
  425. }
  426. var network string
  427. if info := infos.Protocols["eth"]; info != nil {
  428. network = fmt.Sprintf("%d", info.(*ethproto.NodeInfo).Network)
  429. } else {
  430. network = fmt.Sprintf("%d", infos.Protocols["les"].(*les.NodeInfo).Network)
  431. }
  432. auth := &authMsg{
  433. ID: s.node,
  434. Info: nodeInfo{
  435. Name: s.node,
  436. Node: infos.Name,
  437. Port: infos.Ports.Listener,
  438. Network: network,
  439. Protocol: strings.Join(protocols, ", "),
  440. API: "No",
  441. Os: runtime.GOOS,
  442. OsVer: runtime.GOARCH,
  443. Client: "0.1.1",
  444. History: true,
  445. },
  446. Secret: s.pass,
  447. }
  448. login := map[string][]interface{}{
  449. "emit": {"hello", auth},
  450. }
  451. if err := conn.WriteJSON(login); err != nil {
  452. return err
  453. }
  454. // Retrieve the remote ack or connection termination
  455. var ack map[string][]string
  456. if err := conn.ReadJSON(&ack); err != nil || len(ack["emit"]) != 1 || ack["emit"][0] != "ready" {
  457. return errors.New("unauthorized")
  458. }
  459. return nil
  460. }
  461. // report collects all possible data to report and send it to the stats server.
  462. // This should only be used on reconnects or rarely to avoid overloading the
  463. // server. Use the individual methods for reporting subscribed events.
  464. func (s *Service) report(conn *connWrapper) error {
  465. if err := s.reportLatency(conn); err != nil {
  466. return err
  467. }
  468. if err := s.reportBlock(conn, nil); err != nil {
  469. return err
  470. }
  471. if err := s.reportPending(conn); err != nil {
  472. return err
  473. }
  474. if err := s.reportStats(conn); err != nil {
  475. return err
  476. }
  477. return nil
  478. }
  479. // reportLatency sends a ping request to the server, measures the RTT time and
  480. // finally sends a latency update.
  481. func (s *Service) reportLatency(conn *connWrapper) error {
  482. // Send the current time to the ethstats server
  483. start := time.Now()
  484. ping := map[string][]interface{}{
  485. "emit": {"node-ping", map[string]string{
  486. "id": s.node,
  487. "clientTime": start.String(),
  488. }},
  489. }
  490. if err := conn.WriteJSON(ping); err != nil {
  491. return err
  492. }
  493. // Wait for the pong request to arrive back
  494. select {
  495. case <-s.pongCh:
  496. // Pong delivered, report the latency
  497. case <-time.After(5 * time.Second):
  498. // Ping timeout, abort
  499. return errors.New("ping timed out")
  500. }
  501. latency := strconv.Itoa(int((time.Since(start) / time.Duration(2)).Nanoseconds() / 1000000))
  502. // Send back the measured latency
  503. log.Trace("Sending measured latency to ethstats", "latency", latency)
  504. stats := map[string][]interface{}{
  505. "emit": {"latency", map[string]string{
  506. "id": s.node,
  507. "latency": latency,
  508. }},
  509. }
  510. return conn.WriteJSON(stats)
  511. }
  512. // blockStats is the information to report about individual blocks.
  513. type blockStats struct {
  514. Number *big.Int `json:"number"`
  515. Hash common.Hash `json:"hash"`
  516. ParentHash common.Hash `json:"parentHash"`
  517. Timestamp *big.Int `json:"timestamp"`
  518. Miner common.Address `json:"miner"`
  519. GasUsed uint64 `json:"gasUsed"`
  520. GasLimit uint64 `json:"gasLimit"`
  521. Diff string `json:"difficulty"`
  522. TotalDiff string `json:"totalDifficulty"`
  523. Txs []txStats `json:"transactions"`
  524. TxHash common.Hash `json:"transactionsRoot"`
  525. Root common.Hash `json:"stateRoot"`
  526. Uncles uncleStats `json:"uncles"`
  527. }
  528. // txStats is the information to report about individual transactions.
  529. type txStats struct {
  530. Hash common.Hash `json:"hash"`
  531. }
  532. // uncleStats is a custom wrapper around an uncle array to force serializing
  533. // empty arrays instead of returning null for them.
  534. type uncleStats []*types.Header
  535. func (s uncleStats) MarshalJSON() ([]byte, error) {
  536. if uncles := ([]*types.Header)(s); len(uncles) > 0 {
  537. return json.Marshal(uncles)
  538. }
  539. return []byte("[]"), nil
  540. }
  541. // reportBlock retrieves the current chain head and reports it to the stats server.
  542. func (s *Service) reportBlock(conn *connWrapper, block *types.Block) error {
  543. // Gather the block details from the header or block chain
  544. details := s.assembleBlockStats(block)
  545. // Assemble the block report and send it to the server
  546. log.Trace("Sending new block to ethstats", "number", details.Number, "hash", details.Hash)
  547. stats := map[string]interface{}{
  548. "id": s.node,
  549. "block": details,
  550. }
  551. report := map[string][]interface{}{
  552. "emit": {"block", stats},
  553. }
  554. return conn.WriteJSON(report)
  555. }
  556. // assembleBlockStats retrieves any required metadata to report a single block
  557. // and assembles the block stats. If block is nil, the current head is processed.
  558. func (s *Service) assembleBlockStats(block *types.Block) *blockStats {
  559. // Gather the block infos from the local blockchain
  560. var (
  561. header *types.Header
  562. td *big.Int
  563. txs []txStats
  564. uncles []*types.Header
  565. )
  566. // check if backend is a full node
  567. fullBackend, ok := s.backend.(fullNodeBackend)
  568. if ok {
  569. if block == nil {
  570. block = fullBackend.CurrentBlock()
  571. }
  572. header = block.Header()
  573. td = fullBackend.GetTd(context.Background(), header.Hash())
  574. txs = make([]txStats, len(block.Transactions()))
  575. for i, tx := range block.Transactions() {
  576. txs[i].Hash = tx.Hash()
  577. }
  578. uncles = block.Uncles()
  579. } else {
  580. // Light nodes would need on-demand lookups for transactions/uncles, skip
  581. if block != nil {
  582. header = block.Header()
  583. } else {
  584. header = s.backend.CurrentHeader()
  585. }
  586. td = s.backend.GetTd(context.Background(), header.Hash())
  587. txs = []txStats{}
  588. }
  589. // Assemble and return the block stats
  590. author, _ := s.engine.Author(header)
  591. return &blockStats{
  592. Number: header.Number,
  593. Hash: header.Hash(),
  594. ParentHash: header.ParentHash,
  595. Timestamp: new(big.Int).SetUint64(header.Time),
  596. Miner: author,
  597. GasUsed: header.GasUsed,
  598. GasLimit: header.GasLimit,
  599. Diff: header.Difficulty.String(),
  600. TotalDiff: td.String(),
  601. Txs: txs,
  602. TxHash: header.TxHash,
  603. Root: header.Root,
  604. Uncles: uncles,
  605. }
  606. }
  607. // reportHistory retrieves the most recent batch of blocks and reports it to the
  608. // stats server.
  609. func (s *Service) reportHistory(conn *connWrapper, list []uint64) error {
  610. // Figure out the indexes that need reporting
  611. indexes := make([]uint64, 0, historyUpdateRange)
  612. if len(list) > 0 {
  613. // Specific indexes requested, send them back in particular
  614. indexes = append(indexes, list...)
  615. } else {
  616. // No indexes requested, send back the top ones
  617. head := s.backend.CurrentHeader().Number.Int64()
  618. start := head - historyUpdateRange + 1
  619. if start < 0 {
  620. start = 0
  621. }
  622. for i := uint64(start); i <= uint64(head); i++ {
  623. indexes = append(indexes, i)
  624. }
  625. }
  626. // Gather the batch of blocks to report
  627. history := make([]*blockStats, len(indexes))
  628. for i, number := range indexes {
  629. fullBackend, ok := s.backend.(fullNodeBackend)
  630. // Retrieve the next block if it's known to us
  631. var block *types.Block
  632. if ok {
  633. block, _ = fullBackend.BlockByNumber(context.Background(), rpc.BlockNumber(number)) // TODO ignore error here ?
  634. } else {
  635. if header, _ := s.backend.HeaderByNumber(context.Background(), rpc.BlockNumber(number)); header != nil {
  636. block = types.NewBlockWithHeader(header)
  637. }
  638. }
  639. // If we do have the block, add to the history and continue
  640. if block != nil {
  641. history[len(history)-1-i] = s.assembleBlockStats(block)
  642. continue
  643. }
  644. // Ran out of blocks, cut the report short and send
  645. history = history[len(history)-i:]
  646. break
  647. }
  648. // Assemble the history report and send it to the server
  649. if len(history) > 0 {
  650. log.Trace("Sending historical blocks to ethstats", "first", history[0].Number, "last", history[len(history)-1].Number)
  651. } else {
  652. log.Trace("No history to send to stats server")
  653. }
  654. stats := map[string]interface{}{
  655. "id": s.node,
  656. "history": history,
  657. }
  658. report := map[string][]interface{}{
  659. "emit": {"history", stats},
  660. }
  661. return conn.WriteJSON(report)
  662. }
  663. // pendStats is the information to report about pending transactions.
  664. type pendStats struct {
  665. Pending int `json:"pending"`
  666. }
  667. // reportPending retrieves the current number of pending transactions and reports
  668. // it to the stats server.
  669. func (s *Service) reportPending(conn *connWrapper) error {
  670. // Retrieve the pending count from the local blockchain
  671. pending, _ := s.backend.Stats()
  672. // Assemble the transaction stats and send it to the server
  673. log.Trace("Sending pending transactions to ethstats", "count", pending)
  674. stats := map[string]interface{}{
  675. "id": s.node,
  676. "stats": &pendStats{
  677. Pending: pending,
  678. },
  679. }
  680. report := map[string][]interface{}{
  681. "emit": {"pending", stats},
  682. }
  683. return conn.WriteJSON(report)
  684. }
  685. // nodeStats is the information to report about the local node.
  686. type nodeStats struct {
  687. Active bool `json:"active"`
  688. Syncing bool `json:"syncing"`
  689. Mining bool `json:"mining"`
  690. Hashrate int `json:"hashrate"`
  691. Peers int `json:"peers"`
  692. GasPrice int `json:"gasPrice"`
  693. Uptime int `json:"uptime"`
  694. }
  695. // reportStats retrieves various stats about the node at the networking and
  696. // mining layer and reports it to the stats server.
  697. func (s *Service) reportStats(conn *connWrapper) error {
  698. // Gather the syncing and mining infos from the local miner instance
  699. var (
  700. mining bool
  701. hashrate int
  702. syncing bool
  703. gasprice int
  704. )
  705. // check if backend is a full node
  706. fullBackend, ok := s.backend.(fullNodeBackend)
  707. if ok {
  708. mining = fullBackend.Miner().Mining()
  709. hashrate = int(fullBackend.Miner().Hashrate())
  710. sync := fullBackend.SyncProgress()
  711. syncing = fullBackend.CurrentHeader().Number.Uint64() >= sync.HighestBlock
  712. price, _ := fullBackend.SuggestGasTipCap(context.Background())
  713. gasprice = int(price.Uint64())
  714. if basefee := fullBackend.CurrentHeader().BaseFee; basefee != nil {
  715. gasprice += int(basefee.Uint64())
  716. }
  717. } else {
  718. sync := s.backend.SyncProgress()
  719. syncing = s.backend.CurrentHeader().Number.Uint64() >= sync.HighestBlock
  720. }
  721. // Assemble the node stats and send it to the server
  722. log.Trace("Sending node details to ethstats")
  723. stats := map[string]interface{}{
  724. "id": s.node,
  725. "stats": &nodeStats{
  726. Active: true,
  727. Mining: mining,
  728. Hashrate: hashrate,
  729. Peers: s.server.PeerCount(),
  730. GasPrice: gasprice,
  731. Syncing: syncing,
  732. Uptime: 100,
  733. },
  734. }
  735. report := map[string][]interface{}{
  736. "emit": {"stats", stats},
  737. }
  738. return conn.WriteJSON(report)
  739. }