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.

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