You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

756 lines
23 KiB

  1. package node
  2. import (
  3. "bytes"
  4. "context"
  5. "errors"
  6. "fmt"
  7. "net"
  8. "net/http"
  9. "github.com/prometheus/client_golang/prometheus"
  10. "github.com/prometheus/client_golang/prometheus/promhttp"
  11. amino "github.com/tendermint/go-amino"
  12. abci "github.com/tendermint/tendermint/abci/types"
  13. "github.com/tendermint/tendermint/crypto/ed25519"
  14. cmn "github.com/tendermint/tendermint/libs/common"
  15. dbm "github.com/tendermint/tendermint/libs/db"
  16. "github.com/tendermint/tendermint/libs/log"
  17. bc "github.com/tendermint/tendermint/blockchain"
  18. cfg "github.com/tendermint/tendermint/config"
  19. cs "github.com/tendermint/tendermint/consensus"
  20. "github.com/tendermint/tendermint/evidence"
  21. mempl "github.com/tendermint/tendermint/mempool"
  22. "github.com/tendermint/tendermint/p2p"
  23. "github.com/tendermint/tendermint/p2p/pex"
  24. "github.com/tendermint/tendermint/privval"
  25. "github.com/tendermint/tendermint/proxy"
  26. rpccore "github.com/tendermint/tendermint/rpc/core"
  27. ctypes "github.com/tendermint/tendermint/rpc/core/types"
  28. grpccore "github.com/tendermint/tendermint/rpc/grpc"
  29. rpc "github.com/tendermint/tendermint/rpc/lib"
  30. rpcserver "github.com/tendermint/tendermint/rpc/lib/server"
  31. sm "github.com/tendermint/tendermint/state"
  32. "github.com/tendermint/tendermint/state/txindex"
  33. "github.com/tendermint/tendermint/state/txindex/kv"
  34. "github.com/tendermint/tendermint/state/txindex/null"
  35. "github.com/tendermint/tendermint/types"
  36. "github.com/tendermint/tendermint/version"
  37. _ "net/http/pprof"
  38. )
  39. //------------------------------------------------------------------------------
  40. // DBContext specifies config information for loading a new DB.
  41. type DBContext struct {
  42. ID string
  43. Config *cfg.Config
  44. }
  45. // DBProvider takes a DBContext and returns an instantiated DB.
  46. type DBProvider func(*DBContext) (dbm.DB, error)
  47. // DefaultDBProvider returns a database using the DBBackend and DBDir
  48. // specified in the ctx.Config.
  49. func DefaultDBProvider(ctx *DBContext) (dbm.DB, error) {
  50. dbType := dbm.DBBackendType(ctx.Config.DBBackend)
  51. return dbm.NewDB(ctx.ID, dbType, ctx.Config.DBDir()), nil
  52. }
  53. // GenesisDocProvider returns a GenesisDoc.
  54. // It allows the GenesisDoc to be pulled from sources other than the
  55. // filesystem, for instance from a distributed key-value store cluster.
  56. type GenesisDocProvider func() (*types.GenesisDoc, error)
  57. // DefaultGenesisDocProviderFunc returns a GenesisDocProvider that loads
  58. // the GenesisDoc from the config.GenesisFile() on the filesystem.
  59. func DefaultGenesisDocProviderFunc(config *cfg.Config) GenesisDocProvider {
  60. return func() (*types.GenesisDoc, error) {
  61. return types.GenesisDocFromFile(config.GenesisFile())
  62. }
  63. }
  64. // NodeProvider takes a config and a logger and returns a ready to go Node.
  65. type NodeProvider func(*cfg.Config, log.Logger) (*Node, error)
  66. // DefaultNewNode returns a Tendermint node with default settings for the
  67. // PrivValidator, ClientCreator, GenesisDoc, and DBProvider.
  68. // It implements NodeProvider.
  69. func DefaultNewNode(config *cfg.Config, logger log.Logger) (*Node, error) {
  70. return NewNode(config,
  71. privval.LoadOrGenFilePV(config.PrivValidatorFile()),
  72. proxy.DefaultClientCreator(config.ProxyApp, config.ABCI, config.DBDir()),
  73. DefaultGenesisDocProviderFunc(config),
  74. DefaultDBProvider,
  75. DefaultMetricsProvider,
  76. logger,
  77. )
  78. }
  79. // MetricsProvider returns a consensus, p2p and mempool Metrics.
  80. type MetricsProvider func() (*cs.Metrics, *p2p.Metrics, *mempl.Metrics)
  81. // DefaultMetricsProvider returns consensus, p2p and mempool Metrics build
  82. // using Prometheus client library.
  83. func DefaultMetricsProvider() (*cs.Metrics, *p2p.Metrics, *mempl.Metrics) {
  84. return cs.PrometheusMetrics(), p2p.PrometheusMetrics(), mempl.PrometheusMetrics()
  85. }
  86. // NopMetricsProvider returns consensus, p2p and mempool Metrics as no-op.
  87. func NopMetricsProvider() (*cs.Metrics, *p2p.Metrics, *mempl.Metrics) {
  88. return cs.NopMetrics(), p2p.NopMetrics(), mempl.NopMetrics()
  89. }
  90. //------------------------------------------------------------------------------
  91. // Node is the highest level interface to a full Tendermint node.
  92. // It includes all configuration information and running services.
  93. type Node struct {
  94. cmn.BaseService
  95. // config
  96. config *cfg.Config
  97. genesisDoc *types.GenesisDoc // initial validator set
  98. privValidator types.PrivValidator // local node's validator key
  99. // network
  100. sw *p2p.Switch // p2p connections
  101. addrBook pex.AddrBook // known peers
  102. // services
  103. eventBus *types.EventBus // pub/sub for services
  104. stateDB dbm.DB
  105. blockStore *bc.BlockStore // store the blockchain to disk
  106. bcReactor *bc.BlockchainReactor // for fast-syncing
  107. mempoolReactor *mempl.MempoolReactor // for gossipping transactions
  108. consensusState *cs.ConsensusState // latest consensus state
  109. consensusReactor *cs.ConsensusReactor // for participating in the consensus
  110. evidencePool *evidence.EvidencePool // tracking evidence
  111. proxyApp proxy.AppConns // connection to the application
  112. rpcListeners []net.Listener // rpc servers
  113. txIndexer txindex.TxIndexer
  114. indexerService *txindex.IndexerService
  115. prometheusSrv *http.Server
  116. }
  117. // NewNode returns a new, ready to go, Tendermint Node.
  118. func NewNode(config *cfg.Config,
  119. privValidator types.PrivValidator,
  120. clientCreator proxy.ClientCreator,
  121. genesisDocProvider GenesisDocProvider,
  122. dbProvider DBProvider,
  123. metricsProvider MetricsProvider,
  124. logger log.Logger) (*Node, error) {
  125. // Get BlockStore
  126. blockStoreDB, err := dbProvider(&DBContext{"blockstore", config})
  127. if err != nil {
  128. return nil, err
  129. }
  130. blockStore := bc.NewBlockStore(blockStoreDB)
  131. // Get State
  132. stateDB, err := dbProvider(&DBContext{"state", config})
  133. if err != nil {
  134. return nil, err
  135. }
  136. // Get genesis doc
  137. // TODO: move to state package?
  138. genDoc, err := loadGenesisDoc(stateDB)
  139. if err != nil {
  140. genDoc, err = genesisDocProvider()
  141. if err != nil {
  142. return nil, err
  143. }
  144. // save genesis doc to prevent a certain class of user errors (e.g. when it
  145. // was changed, accidentally or not). Also good for audit trail.
  146. saveGenesisDoc(stateDB, genDoc)
  147. }
  148. state, err := sm.LoadStateFromDBOrGenesisDoc(stateDB, genDoc)
  149. if err != nil {
  150. return nil, err
  151. }
  152. // Create the proxyApp, which manages connections (consensus, mempool, query)
  153. // and sync tendermint and the app by performing a handshake
  154. // and replaying any necessary blocks
  155. consensusLogger := logger.With("module", "consensus")
  156. handshaker := cs.NewHandshaker(stateDB, state, blockStore, genDoc)
  157. handshaker.SetLogger(consensusLogger)
  158. proxyApp := proxy.NewAppConns(clientCreator, handshaker)
  159. proxyApp.SetLogger(logger.With("module", "proxy"))
  160. if err := proxyApp.Start(); err != nil {
  161. return nil, fmt.Errorf("Error starting proxy app connections: %v", err)
  162. }
  163. // reload the state (it may have been updated by the handshake)
  164. state = sm.LoadState(stateDB)
  165. // If an address is provided, listen on the socket for a
  166. // connection from an external signing process.
  167. if config.PrivValidatorListenAddr != "" {
  168. var (
  169. // TODO: persist this key so external signer
  170. // can actually authenticate us
  171. privKey = ed25519.GenPrivKey()
  172. pvsc = privval.NewSocketPV(
  173. logger.With("module", "privval"),
  174. config.PrivValidatorListenAddr,
  175. privKey,
  176. )
  177. )
  178. if err := pvsc.Start(); err != nil {
  179. return nil, fmt.Errorf("Error starting private validator client: %v", err)
  180. }
  181. privValidator = pvsc
  182. }
  183. // Decide whether to fast-sync or not
  184. // We don't fast-sync when the only validator is us.
  185. fastSync := config.FastSync
  186. if state.Validators.Size() == 1 {
  187. addr, _ := state.Validators.GetByIndex(0)
  188. if bytes.Equal(privValidator.GetAddress(), addr) {
  189. fastSync = false
  190. }
  191. }
  192. // Log whether this node is a validator or an observer
  193. if state.Validators.HasAddress(privValidator.GetAddress()) {
  194. consensusLogger.Info("This node is a validator", "addr", privValidator.GetAddress(), "pubKey", privValidator.GetPubKey())
  195. } else {
  196. consensusLogger.Info("This node is not a validator", "addr", privValidator.GetAddress(), "pubKey", privValidator.GetPubKey())
  197. }
  198. // metrics
  199. var (
  200. csMetrics *cs.Metrics
  201. p2pMetrics *p2p.Metrics
  202. memplMetrics *mempl.Metrics
  203. )
  204. if config.Instrumentation.Prometheus {
  205. csMetrics, p2pMetrics, memplMetrics = metricsProvider()
  206. } else {
  207. csMetrics, p2pMetrics, memplMetrics = NopMetricsProvider()
  208. }
  209. // Make MempoolReactor
  210. mempoolLogger := logger.With("module", "mempool")
  211. mempool := mempl.NewMempool(
  212. config.Mempool,
  213. proxyApp.Mempool(),
  214. state.LastBlockHeight,
  215. mempl.WithMetrics(memplMetrics),
  216. )
  217. mempool.SetLogger(mempoolLogger)
  218. mempool.InitWAL() // no need to have the mempool wal during tests
  219. mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool)
  220. mempoolReactor.SetLogger(mempoolLogger)
  221. if config.Consensus.WaitForTxs() {
  222. mempool.EnableTxsAvailable()
  223. }
  224. // Make Evidence Reactor
  225. evidenceDB, err := dbProvider(&DBContext{"evidence", config})
  226. if err != nil {
  227. return nil, err
  228. }
  229. evidenceLogger := logger.With("module", "evidence")
  230. evidenceStore := evidence.NewEvidenceStore(evidenceDB)
  231. evidencePool := evidence.NewEvidencePool(stateDB, evidenceStore)
  232. evidencePool.SetLogger(evidenceLogger)
  233. evidenceReactor := evidence.NewEvidenceReactor(evidencePool)
  234. evidenceReactor.SetLogger(evidenceLogger)
  235. blockExecLogger := logger.With("module", "state")
  236. // make block executor for consensus and blockchain reactors to execute blocks
  237. blockExec := sm.NewBlockExecutor(stateDB, blockExecLogger, proxyApp.Consensus(), mempool, evidencePool)
  238. // Make BlockchainReactor
  239. bcReactor := bc.NewBlockchainReactor(state.Copy(), blockExec, blockStore, fastSync)
  240. bcReactor.SetLogger(logger.With("module", "blockchain"))
  241. // Make ConsensusReactor
  242. consensusState := cs.NewConsensusState(
  243. config.Consensus,
  244. state.Copy(),
  245. blockExec,
  246. blockStore,
  247. mempool,
  248. evidencePool,
  249. cs.WithMetrics(csMetrics),
  250. )
  251. consensusState.SetLogger(consensusLogger)
  252. if privValidator != nil {
  253. consensusState.SetPrivValidator(privValidator)
  254. }
  255. consensusReactor := cs.NewConsensusReactor(consensusState, fastSync)
  256. consensusReactor.SetLogger(consensusLogger)
  257. p2pLogger := logger.With("module", "p2p")
  258. sw := p2p.NewSwitch(config.P2P, p2p.WithMetrics(p2pMetrics))
  259. sw.SetLogger(p2pLogger)
  260. sw.AddReactor("MEMPOOL", mempoolReactor)
  261. sw.AddReactor("BLOCKCHAIN", bcReactor)
  262. sw.AddReactor("CONSENSUS", consensusReactor)
  263. sw.AddReactor("EVIDENCE", evidenceReactor)
  264. // Optionally, start the pex reactor
  265. //
  266. // TODO:
  267. //
  268. // We need to set Seeds and PersistentPeers on the switch,
  269. // since it needs to be able to use these (and their DNS names)
  270. // even if the PEX is off. We can include the DNS name in the NetAddress,
  271. // but it would still be nice to have a clear list of the current "PersistentPeers"
  272. // somewhere that we can return with net_info.
  273. //
  274. // If PEX is on, it should handle dialing the seeds. Otherwise the switch does it.
  275. // Note we currently use the addrBook regardless at least for AddOurAddress
  276. addrBook := pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
  277. addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
  278. if config.P2P.PexReactor {
  279. // TODO persistent peers ? so we can have their DNS addrs saved
  280. pexReactor := pex.NewPEXReactor(addrBook,
  281. &pex.PEXReactorConfig{
  282. Seeds: cmn.SplitAndTrim(config.P2P.Seeds, ",", " "),
  283. SeedMode: config.P2P.SeedMode,
  284. })
  285. pexReactor.SetLogger(p2pLogger)
  286. sw.AddReactor("PEX", pexReactor)
  287. }
  288. sw.SetAddrBook(addrBook)
  289. // Filter peers by addr or pubkey with an ABCI query.
  290. // If the query return code is OK, add peer.
  291. // XXX: Query format subject to change
  292. if config.FilterPeers {
  293. // NOTE: addr is ip:port
  294. sw.SetAddrFilter(func(addr net.Addr) error {
  295. resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/addr/%s", addr.String())})
  296. if err != nil {
  297. return err
  298. }
  299. if resQuery.IsErr() {
  300. return fmt.Errorf("Error querying abci app: %v", resQuery)
  301. }
  302. return nil
  303. })
  304. sw.SetIDFilter(func(id p2p.ID) error {
  305. resQuery, err := proxyApp.Query().QuerySync(abci.RequestQuery{Path: cmn.Fmt("/p2p/filter/id/%s", id)})
  306. if err != nil {
  307. return err
  308. }
  309. if resQuery.IsErr() {
  310. return fmt.Errorf("Error querying abci app: %v", resQuery)
  311. }
  312. return nil
  313. })
  314. }
  315. eventBus := types.NewEventBus()
  316. eventBus.SetLogger(logger.With("module", "events"))
  317. // services which will be publishing and/or subscribing for messages (events)
  318. // consensusReactor will set it on consensusState and blockExecutor
  319. consensusReactor.SetEventBus(eventBus)
  320. // Transaction indexing
  321. var txIndexer txindex.TxIndexer
  322. switch config.TxIndex.Indexer {
  323. case "kv":
  324. store, err := dbProvider(&DBContext{"tx_index", config})
  325. if err != nil {
  326. return nil, err
  327. }
  328. if config.TxIndex.IndexTags != "" {
  329. txIndexer = kv.NewTxIndex(store, kv.IndexTags(cmn.SplitAndTrim(config.TxIndex.IndexTags, ",", " ")))
  330. } else if config.TxIndex.IndexAllTags {
  331. txIndexer = kv.NewTxIndex(store, kv.IndexAllTags())
  332. } else {
  333. txIndexer = kv.NewTxIndex(store)
  334. }
  335. default:
  336. txIndexer = &null.TxIndex{}
  337. }
  338. indexerService := txindex.NewIndexerService(txIndexer, eventBus)
  339. indexerService.SetLogger(logger.With("module", "txindex"))
  340. // run the profile server
  341. profileHost := config.ProfListenAddress
  342. if profileHost != "" {
  343. go func() {
  344. logger.Error("Profile server", "err", http.ListenAndServe(profileHost, nil))
  345. }()
  346. }
  347. node := &Node{
  348. config: config,
  349. genesisDoc: genDoc,
  350. privValidator: privValidator,
  351. sw: sw,
  352. addrBook: addrBook,
  353. stateDB: stateDB,
  354. blockStore: blockStore,
  355. bcReactor: bcReactor,
  356. mempoolReactor: mempoolReactor,
  357. consensusState: consensusState,
  358. consensusReactor: consensusReactor,
  359. evidencePool: evidencePool,
  360. proxyApp: proxyApp,
  361. txIndexer: txIndexer,
  362. indexerService: indexerService,
  363. eventBus: eventBus,
  364. }
  365. node.BaseService = *cmn.NewBaseService(logger, "Node", node)
  366. return node, nil
  367. }
  368. // OnStart starts the Node. It implements cmn.Service.
  369. func (n *Node) OnStart() error {
  370. err := n.eventBus.Start()
  371. if err != nil {
  372. return err
  373. }
  374. // Create & add listener
  375. l := p2p.NewDefaultListener(
  376. n.config.P2P.ListenAddress,
  377. n.config.P2P.ExternalAddress,
  378. n.config.P2P.UPNP,
  379. n.Logger.With("module", "p2p"))
  380. n.sw.AddListener(l)
  381. // Generate node PrivKey
  382. // TODO: pass in like privValidator
  383. nodeKey, err := p2p.LoadOrGenNodeKey(n.config.NodeKeyFile())
  384. if err != nil {
  385. return err
  386. }
  387. n.Logger.Info("P2P Node ID", "ID", nodeKey.ID(), "file", n.config.NodeKeyFile())
  388. nodeInfo := n.makeNodeInfo(nodeKey.ID())
  389. n.sw.SetNodeInfo(nodeInfo)
  390. n.sw.SetNodeKey(nodeKey)
  391. // Add ourselves to addrbook to prevent dialing ourselves
  392. n.addrBook.AddOurAddress(nodeInfo.NetAddress())
  393. // Add private IDs to addrbook to block those peers being added
  394. n.addrBook.AddPrivateIDs(cmn.SplitAndTrim(n.config.P2P.PrivatePeerIDs, ",", " "))
  395. // Start the RPC server before the P2P server
  396. // so we can eg. receive txs for the first block
  397. if n.config.RPC.ListenAddress != "" {
  398. listeners, err := n.startRPC()
  399. if err != nil {
  400. return err
  401. }
  402. n.rpcListeners = listeners
  403. }
  404. if n.config.Instrumentation.Prometheus {
  405. n.prometheusSrv = n.startPrometheusServer(n.config.Instrumentation.PrometheusListenAddr)
  406. }
  407. // Start the switch (the P2P server).
  408. err = n.sw.Start()
  409. if err != nil {
  410. return err
  411. }
  412. // Always connect to persistent peers
  413. if n.config.P2P.PersistentPeers != "" {
  414. err = n.sw.DialPeersAsync(n.addrBook, cmn.SplitAndTrim(n.config.P2P.PersistentPeers, ",", " "), true)
  415. if err != nil {
  416. return err
  417. }
  418. }
  419. // start tx indexer
  420. return n.indexerService.Start()
  421. }
  422. // OnStop stops the Node. It implements cmn.Service.
  423. func (n *Node) OnStop() {
  424. n.BaseService.OnStop()
  425. n.Logger.Info("Stopping Node")
  426. // first stop the non-reactor services
  427. n.eventBus.Stop()
  428. n.indexerService.Stop()
  429. // now stop the reactors
  430. // TODO: gracefully disconnect from peers.
  431. n.sw.Stop()
  432. // finally stop the listeners / external services
  433. for _, l := range n.rpcListeners {
  434. n.Logger.Info("Closing rpc listener", "listener", l)
  435. if err := l.Close(); err != nil {
  436. n.Logger.Error("Error closing listener", "listener", l, "err", err)
  437. }
  438. }
  439. if pvsc, ok := n.privValidator.(*privval.SocketPV); ok {
  440. if err := pvsc.Stop(); err != nil {
  441. n.Logger.Error("Error stopping priv validator socket client", "err", err)
  442. }
  443. }
  444. if n.prometheusSrv != nil {
  445. if err := n.prometheusSrv.Shutdown(context.Background()); err != nil {
  446. // Error from closing listeners, or context timeout:
  447. n.Logger.Error("Prometheus HTTP server Shutdown", "err", err)
  448. }
  449. }
  450. }
  451. // RunForever waits for an interrupt signal and stops the node.
  452. func (n *Node) RunForever() {
  453. // Sleep forever and then...
  454. cmn.TrapSignal(func() {
  455. n.Stop()
  456. })
  457. }
  458. // AddListener adds a listener to accept inbound peer connections.
  459. // It should be called before starting the Node.
  460. // The first listener is the primary listener (in NodeInfo)
  461. func (n *Node) AddListener(l p2p.Listener) {
  462. n.sw.AddListener(l)
  463. }
  464. // ConfigureRPC sets all variables in rpccore so they will serve
  465. // rpc calls from this node
  466. func (n *Node) ConfigureRPC() {
  467. rpccore.SetStateDB(n.stateDB)
  468. rpccore.SetBlockStore(n.blockStore)
  469. rpccore.SetConsensusState(n.consensusState)
  470. rpccore.SetMempool(n.mempoolReactor.Mempool)
  471. rpccore.SetEvidencePool(n.evidencePool)
  472. rpccore.SetSwitch(n.sw)
  473. rpccore.SetPubKey(n.privValidator.GetPubKey())
  474. rpccore.SetGenesisDoc(n.genesisDoc)
  475. rpccore.SetAddrBook(n.addrBook)
  476. rpccore.SetProxyAppQuery(n.proxyApp.Query())
  477. rpccore.SetTxIndexer(n.txIndexer)
  478. rpccore.SetConsensusReactor(n.consensusReactor)
  479. rpccore.SetEventBus(n.eventBus)
  480. rpccore.SetLogger(n.Logger.With("module", "rpc"))
  481. }
  482. func (n *Node) startRPC() ([]net.Listener, error) {
  483. n.ConfigureRPC()
  484. listenAddrs := cmn.SplitAndTrim(n.config.RPC.ListenAddress, ",", " ")
  485. coreCodec := amino.NewCodec()
  486. ctypes.RegisterAmino(coreCodec)
  487. if n.config.RPC.Unsafe {
  488. rpccore.AddUnsafeRoutes()
  489. }
  490. // we may expose the rpc over both a unix and tcp socket
  491. listeners := make([]net.Listener, len(listenAddrs))
  492. for i, listenAddr := range listenAddrs {
  493. mux := http.NewServeMux()
  494. rpcLogger := n.Logger.With("module", "rpc-server")
  495. wm := rpcserver.NewWebsocketManager(rpccore.Routes, coreCodec, rpcserver.EventSubscriber(n.eventBus))
  496. wm.SetLogger(rpcLogger.With("protocol", "websocket"))
  497. mux.HandleFunc("/websocket", wm.WebsocketHandler)
  498. rpcserver.RegisterRPCFuncs(mux, rpccore.Routes, coreCodec, rpcLogger)
  499. listener, err := rpcserver.StartHTTPServer(
  500. listenAddr,
  501. mux,
  502. rpcLogger,
  503. rpcserver.Config{MaxOpenConnections: n.config.RPC.MaxOpenConnections},
  504. )
  505. if err != nil {
  506. return nil, err
  507. }
  508. listeners[i] = listener
  509. }
  510. // we expose a simplified api over grpc for convenience to app devs
  511. grpcListenAddr := n.config.RPC.GRPCListenAddress
  512. if grpcListenAddr != "" {
  513. listener, err := grpccore.StartGRPCServer(
  514. grpcListenAddr,
  515. grpccore.Config{
  516. MaxOpenConnections: n.config.RPC.GRPCMaxOpenConnections,
  517. },
  518. )
  519. if err != nil {
  520. return nil, err
  521. }
  522. listeners = append(listeners, listener)
  523. }
  524. return listeners, nil
  525. }
  526. // startPrometheusServer starts a Prometheus HTTP server, listening for metrics
  527. // collectors on addr.
  528. func (n *Node) startPrometheusServer(addr string) *http.Server {
  529. srv := &http.Server{
  530. Addr: addr,
  531. Handler: promhttp.InstrumentMetricHandler(
  532. prometheus.DefaultRegisterer, promhttp.HandlerFor(
  533. prometheus.DefaultGatherer,
  534. promhttp.HandlerOpts{MaxRequestsInFlight: n.config.Instrumentation.MaxOpenConnections},
  535. ),
  536. ),
  537. }
  538. go func() {
  539. if err := srv.ListenAndServe(); err != http.ErrServerClosed {
  540. // Error starting or closing listener:
  541. n.Logger.Error("Prometheus HTTP server ListenAndServe", "err", err)
  542. }
  543. }()
  544. return srv
  545. }
  546. // Switch returns the Node's Switch.
  547. func (n *Node) Switch() *p2p.Switch {
  548. return n.sw
  549. }
  550. // BlockStore returns the Node's BlockStore.
  551. func (n *Node) BlockStore() *bc.BlockStore {
  552. return n.blockStore
  553. }
  554. // ConsensusState returns the Node's ConsensusState.
  555. func (n *Node) ConsensusState() *cs.ConsensusState {
  556. return n.consensusState
  557. }
  558. // ConsensusReactor returns the Node's ConsensusReactor.
  559. func (n *Node) ConsensusReactor() *cs.ConsensusReactor {
  560. return n.consensusReactor
  561. }
  562. // MempoolReactor returns the Node's MempoolReactor.
  563. func (n *Node) MempoolReactor() *mempl.MempoolReactor {
  564. return n.mempoolReactor
  565. }
  566. // EvidencePool returns the Node's EvidencePool.
  567. func (n *Node) EvidencePool() *evidence.EvidencePool {
  568. return n.evidencePool
  569. }
  570. // EventBus returns the Node's EventBus.
  571. func (n *Node) EventBus() *types.EventBus {
  572. return n.eventBus
  573. }
  574. // PrivValidator returns the Node's PrivValidator.
  575. // XXX: for convenience only!
  576. func (n *Node) PrivValidator() types.PrivValidator {
  577. return n.privValidator
  578. }
  579. // GenesisDoc returns the Node's GenesisDoc.
  580. func (n *Node) GenesisDoc() *types.GenesisDoc {
  581. return n.genesisDoc
  582. }
  583. // ProxyApp returns the Node's AppConns, representing its connections to the ABCI application.
  584. func (n *Node) ProxyApp() proxy.AppConns {
  585. return n.proxyApp
  586. }
  587. func (n *Node) makeNodeInfo(nodeID p2p.ID) p2p.NodeInfo {
  588. txIndexerStatus := "on"
  589. if _, ok := n.txIndexer.(*null.TxIndex); ok {
  590. txIndexerStatus = "off"
  591. }
  592. nodeInfo := p2p.NodeInfo{
  593. ID: nodeID,
  594. Network: n.genesisDoc.ChainID,
  595. Version: version.Version,
  596. Channels: []byte{
  597. bc.BlockchainChannel,
  598. cs.StateChannel, cs.DataChannel, cs.VoteChannel, cs.VoteSetBitsChannel,
  599. mempl.MempoolChannel,
  600. evidence.EvidenceChannel,
  601. },
  602. Moniker: n.config.Moniker,
  603. Other: []string{
  604. cmn.Fmt("amino_version=%v", amino.Version),
  605. cmn.Fmt("p2p_version=%v", p2p.Version),
  606. cmn.Fmt("consensus_version=%v", cs.Version),
  607. cmn.Fmt("rpc_version=%v/%v", rpc.Version, rpccore.Version),
  608. cmn.Fmt("tx_index=%v", txIndexerStatus),
  609. },
  610. }
  611. if n.config.P2P.PexReactor {
  612. nodeInfo.Channels = append(nodeInfo.Channels, pex.PexChannel)
  613. }
  614. rpcListenAddr := n.config.RPC.ListenAddress
  615. nodeInfo.Other = append(nodeInfo.Other, cmn.Fmt("rpc_addr=%v", rpcListenAddr))
  616. if !n.sw.IsListening() {
  617. return nodeInfo
  618. }
  619. p2pListener := n.sw.Listeners()[0]
  620. p2pHost := p2pListener.ExternalAddressHost()
  621. p2pPort := p2pListener.ExternalAddress().Port
  622. nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
  623. return nodeInfo
  624. }
  625. //------------------------------------------------------------------------------
  626. // NodeInfo returns the Node's Info from the Switch.
  627. func (n *Node) NodeInfo() p2p.NodeInfo {
  628. return n.sw.NodeInfo()
  629. }
  630. //------------------------------------------------------------------------------
  631. var (
  632. genesisDocKey = []byte("genesisDoc")
  633. )
  634. // panics if failed to unmarshal bytes
  635. func loadGenesisDoc(db dbm.DB) (*types.GenesisDoc, error) {
  636. bytes := db.Get(genesisDocKey)
  637. if len(bytes) == 0 {
  638. return nil, errors.New("Genesis doc not found")
  639. }
  640. var genDoc *types.GenesisDoc
  641. err := cdc.UnmarshalJSON(bytes, &genDoc)
  642. if err != nil {
  643. cmn.PanicCrisis(fmt.Sprintf("Failed to load genesis doc due to unmarshaling error: %v (bytes: %X)", err, bytes))
  644. }
  645. return genDoc, nil
  646. }
  647. // panics if failed to marshal the given genesis document
  648. func saveGenesisDoc(db dbm.DB, genDoc *types.GenesisDoc) {
  649. bytes, err := cdc.MarshalJSON(genDoc)
  650. if err != nil {
  651. cmn.PanicCrisis(fmt.Sprintf("Failed to save genesis doc due to marshaling error: %v", err))
  652. }
  653. db.SetSync(genesisDocKey, bytes)
  654. }