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.

929 lines
27 KiB

max-bytes PR follow-up (#2318) * ReapMaxTxs: return all txs if max is negative this mirrors ReapMaxBytes behavior See https://github.com/tendermint/tendermint/pull/2184#discussion_r214439950 * increase MaxAminoOverheadForBlock tested with: ``` func TestMaxAminoOverheadForBlock(t *testing.T) { maxChainID := "" for i := 0; i < MaxChainIDLen; i++ { maxChainID += "𠜎" } h := Header{ ChainID: maxChainID, Height: 10, Time: time.Now().UTC(), NumTxs: 100, TotalTxs: 200, LastBlockID: makeBlockID(make([]byte, 20), 300, make([]byte, 20)), LastCommitHash: tmhash.Sum([]byte("last_commit_hash")), DataHash: tmhash.Sum([]byte("data_hash")), ValidatorsHash: tmhash.Sum([]byte("validators_hash")), NextValidatorsHash: tmhash.Sum([]byte("next_validators_hash")), ConsensusHash: tmhash.Sum([]byte("consensus_hash")), AppHash: tmhash.Sum([]byte("app_hash")), LastResultsHash: tmhash.Sum([]byte("last_results_hash")), EvidenceHash: tmhash.Sum([]byte("evidence_hash")), ProposerAddress: tmhash.Sum([]byte("proposer_address")), } b := Block{ Header: h, Data: Data{Txs: makeTxs(10000, 100)}, Evidence: EvidenceData{}, LastCommit: &Commit{}, } bz, err := cdc.MarshalBinary(b) require.NoError(t, err) assert.Equal(t, MaxHeaderBytes+MaxAminoOverheadForBlock-2, len(bz)-1000000-20000-1) } ``` * fix MaxYYY constants calculation by using math.MaxInt64 See https://github.com/tendermint/tendermint/pull/2184#discussion_r214444244 * pass mempool filter as an option See https://github.com/tendermint/tendermint/pull/2184#discussion_r214445869 * fixes after Dev's comments
6 years ago
Close and retry a RemoteSigner on err (#2923) * Close and recreate a RemoteSigner on err * Update changelog * Address Anton's comments / suggestions: - update changelog - restart TCPVal - shut down on `ErrUnexpectedResponse` * re-init remote signer client with fresh connection if Ping fails - add/update TODOs in secret connection - rename tcp.go -> tcp_client.go, same with ipc to clarify their purpose * account for `conn returned by waitConnection can be `nil` - also add TODO about RemoteSigner conn field * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn - add rwmutex for conn field in IPC * comments and doc.go * fix ipc tests. fixes #2677 * use constants for tests * cleanup some error statements * fixes #2784, race in tests * remove print statement * minor fixes from review * update comment on sts spec * cosmetics * p2p/conn: add failing tests * p2p/conn: make SecretConnection thread safe * changelog * IPCVal signer refactor - use a .reset() method - don't use embedded RemoteSignerClient - guard RemoteSignerClient with mutex - drop the .conn - expose Close() on RemoteSignerClient * apply IPCVal refactor to TCPVal * remove mtx from RemoteSignerClient * consolidate IPCVal and TCPVal, fixes #3104 - done in tcp_client.go - now called SocketVal - takes a listener in the constructor - make tcpListener and unixListener contain all the differences * delete ipc files * introduce unix and tcp dialer for RemoteSigner * rename files - drop tcp_ prefix - rename priv_validator.go to file.go * bring back listener options * fix node * fix priv_val_server * fix node test * minor cleanup and comments
6 years ago
Close and retry a RemoteSigner on err (#2923) * Close and recreate a RemoteSigner on err * Update changelog * Address Anton's comments / suggestions: - update changelog - restart TCPVal - shut down on `ErrUnexpectedResponse` * re-init remote signer client with fresh connection if Ping fails - add/update TODOs in secret connection - rename tcp.go -> tcp_client.go, same with ipc to clarify their purpose * account for `conn returned by waitConnection can be `nil` - also add TODO about RemoteSigner conn field * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn - add rwmutex for conn field in IPC * comments and doc.go * fix ipc tests. fixes #2677 * use constants for tests * cleanup some error statements * fixes #2784, race in tests * remove print statement * minor fixes from review * update comment on sts spec * cosmetics * p2p/conn: add failing tests * p2p/conn: make SecretConnection thread safe * changelog * IPCVal signer refactor - use a .reset() method - don't use embedded RemoteSignerClient - guard RemoteSignerClient with mutex - drop the .conn - expose Close() on RemoteSignerClient * apply IPCVal refactor to TCPVal * remove mtx from RemoteSignerClient * consolidate IPCVal and TCPVal, fixes #3104 - done in tcp_client.go - now called SocketVal - takes a listener in the constructor - make tcpListener and unixListener contain all the differences * delete ipc files * introduce unix and tcp dialer for RemoteSigner * rename files - drop tcp_ prefix - rename priv_validator.go to file.go * bring back listener options * fix node * fix priv_val_server * fix node test * minor cleanup and comments
6 years ago
Close and retry a RemoteSigner on err (#2923) * Close and recreate a RemoteSigner on err * Update changelog * Address Anton's comments / suggestions: - update changelog - restart TCPVal - shut down on `ErrUnexpectedResponse` * re-init remote signer client with fresh connection if Ping fails - add/update TODOs in secret connection - rename tcp.go -> tcp_client.go, same with ipc to clarify their purpose * account for `conn returned by waitConnection can be `nil` - also add TODO about RemoteSigner conn field * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn - add rwmutex for conn field in IPC * comments and doc.go * fix ipc tests. fixes #2677 * use constants for tests * cleanup some error statements * fixes #2784, race in tests * remove print statement * minor fixes from review * update comment on sts spec * cosmetics * p2p/conn: add failing tests * p2p/conn: make SecretConnection thread safe * changelog * IPCVal signer refactor - use a .reset() method - don't use embedded RemoteSignerClient - guard RemoteSignerClient with mutex - drop the .conn - expose Close() on RemoteSignerClient * apply IPCVal refactor to TCPVal * remove mtx from RemoteSignerClient * consolidate IPCVal and TCPVal, fixes #3104 - done in tcp_client.go - now called SocketVal - takes a listener in the constructor - make tcpListener and unixListener contain all the differences * delete ipc files * introduce unix and tcp dialer for RemoteSigner * rename files - drop tcp_ prefix - rename priv_validator.go to file.go * bring back listener options * fix node * fix priv_val_server * fix node test * minor cleanup and comments
6 years ago
Close and retry a RemoteSigner on err (#2923) * Close and recreate a RemoteSigner on err * Update changelog * Address Anton's comments / suggestions: - update changelog - restart TCPVal - shut down on `ErrUnexpectedResponse` * re-init remote signer client with fresh connection if Ping fails - add/update TODOs in secret connection - rename tcp.go -> tcp_client.go, same with ipc to clarify their purpose * account for `conn returned by waitConnection can be `nil` - also add TODO about RemoteSigner conn field * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn - add rwmutex for conn field in IPC * comments and doc.go * fix ipc tests. fixes #2677 * use constants for tests * cleanup some error statements * fixes #2784, race in tests * remove print statement * minor fixes from review * update comment on sts spec * cosmetics * p2p/conn: add failing tests * p2p/conn: make SecretConnection thread safe * changelog * IPCVal signer refactor - use a .reset() method - don't use embedded RemoteSignerClient - guard RemoteSignerClient with mutex - drop the .conn - expose Close() on RemoteSignerClient * apply IPCVal refactor to TCPVal * remove mtx from RemoteSignerClient * consolidate IPCVal and TCPVal, fixes #3104 - done in tcp_client.go - now called SocketVal - takes a listener in the constructor - make tcpListener and unixListener contain all the differences * delete ipc files * introduce unix and tcp dialer for RemoteSigner * rename files - drop tcp_ prefix - rename priv_validator.go to file.go * bring back listener options * fix node * fix priv_val_server * fix node test * minor cleanup and comments
6 years ago
Close and retry a RemoteSigner on err (#2923) * Close and recreate a RemoteSigner on err * Update changelog * Address Anton's comments / suggestions: - update changelog - restart TCPVal - shut down on `ErrUnexpectedResponse` * re-init remote signer client with fresh connection if Ping fails - add/update TODOs in secret connection - rename tcp.go -> tcp_client.go, same with ipc to clarify their purpose * account for `conn returned by waitConnection can be `nil` - also add TODO about RemoteSigner conn field * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn * Tests for retrying: IPC / TCP - shorter info log on success - set conn and use it in tests to close conn - add rwmutex for conn field in IPC * comments and doc.go * fix ipc tests. fixes #2677 * use constants for tests * cleanup some error statements * fixes #2784, race in tests * remove print statement * minor fixes from review * update comment on sts spec * cosmetics * p2p/conn: add failing tests * p2p/conn: make SecretConnection thread safe * changelog * IPCVal signer refactor - use a .reset() method - don't use embedded RemoteSignerClient - guard RemoteSignerClient with mutex - drop the .conn - expose Close() on RemoteSignerClient * apply IPCVal refactor to TCPVal * remove mtx from RemoteSignerClient * consolidate IPCVal and TCPVal, fixes #3104 - done in tcp_client.go - now called SocketVal - takes a listener in the constructor - make tcpListener and unixListener contain all the differences * delete ipc files * introduce unix and tcp dialer for RemoteSigner * rename files - drop tcp_ prefix - rename priv_validator.go to file.go * bring back listener options * fix node * fix priv_val_server * fix node test * minor cleanup and comments
6 years ago
  1. package node
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "net"
  7. "net/http"
  8. _ "net/http/pprof"
  9. "os"
  10. "strings"
  11. "time"
  12. "github.com/pkg/errors"
  13. "github.com/prometheus/client_golang/prometheus"
  14. "github.com/prometheus/client_golang/prometheus/promhttp"
  15. "github.com/rs/cors"
  16. amino "github.com/tendermint/go-amino"
  17. abci "github.com/tendermint/tendermint/abci/types"
  18. bc "github.com/tendermint/tendermint/blockchain"
  19. cfg "github.com/tendermint/tendermint/config"
  20. cs "github.com/tendermint/tendermint/consensus"
  21. "github.com/tendermint/tendermint/crypto/ed25519"
  22. "github.com/tendermint/tendermint/evidence"
  23. cmn "github.com/tendermint/tendermint/libs/common"
  24. dbm "github.com/tendermint/tendermint/libs/db"
  25. "github.com/tendermint/tendermint/libs/log"
  26. mempl "github.com/tendermint/tendermint/mempool"
  27. "github.com/tendermint/tendermint/p2p"
  28. "github.com/tendermint/tendermint/p2p/pex"
  29. "github.com/tendermint/tendermint/privval"
  30. "github.com/tendermint/tendermint/proxy"
  31. rpccore "github.com/tendermint/tendermint/rpc/core"
  32. ctypes "github.com/tendermint/tendermint/rpc/core/types"
  33. grpccore "github.com/tendermint/tendermint/rpc/grpc"
  34. "github.com/tendermint/tendermint/rpc/lib/server"
  35. sm "github.com/tendermint/tendermint/state"
  36. "github.com/tendermint/tendermint/state/txindex"
  37. "github.com/tendermint/tendermint/state/txindex/kv"
  38. "github.com/tendermint/tendermint/state/txindex/null"
  39. "github.com/tendermint/tendermint/types"
  40. tmtime "github.com/tendermint/tendermint/types/time"
  41. "github.com/tendermint/tendermint/version"
  42. )
  43. //------------------------------------------------------------------------------
  44. // DBContext specifies config information for loading a new DB.
  45. type DBContext struct {
  46. ID string
  47. Config *cfg.Config
  48. }
  49. // DBProvider takes a DBContext and returns an instantiated DB.
  50. type DBProvider func(*DBContext) (dbm.DB, error)
  51. // DefaultDBProvider returns a database using the DBBackend and DBDir
  52. // specified in the ctx.Config.
  53. func DefaultDBProvider(ctx *DBContext) (dbm.DB, error) {
  54. dbType := dbm.DBBackendType(ctx.Config.DBBackend)
  55. return dbm.NewDB(ctx.ID, dbType, ctx.Config.DBDir()), nil
  56. }
  57. // GenesisDocProvider returns a GenesisDoc.
  58. // It allows the GenesisDoc to be pulled from sources other than the
  59. // filesystem, for instance from a distributed key-value store cluster.
  60. type GenesisDocProvider func() (*types.GenesisDoc, error)
  61. // DefaultGenesisDocProviderFunc returns a GenesisDocProvider that loads
  62. // the GenesisDoc from the config.GenesisFile() on the filesystem.
  63. func DefaultGenesisDocProviderFunc(config *cfg.Config) GenesisDocProvider {
  64. return func() (*types.GenesisDoc, error) {
  65. return types.GenesisDocFromFile(config.GenesisFile())
  66. }
  67. }
  68. // NodeProvider takes a config and a logger and returns a ready to go Node.
  69. type NodeProvider func(*cfg.Config, log.Logger) (*Node, error)
  70. // DefaultNewNode returns a Tendermint node with default settings for the
  71. // PrivValidator, ClientCreator, GenesisDoc, and DBProvider.
  72. // It implements NodeProvider.
  73. func DefaultNewNode(config *cfg.Config, logger log.Logger) (*Node, error) {
  74. // Generate node PrivKey
  75. nodeKey, err := p2p.LoadOrGenNodeKey(config.NodeKeyFile())
  76. if err != nil {
  77. return nil, err
  78. }
  79. // Convert old PrivValidator if it exists.
  80. oldPrivVal := config.OldPrivValidatorFile()
  81. newPrivValKey := config.PrivValidatorKeyFile()
  82. newPrivValState := config.PrivValidatorStateFile()
  83. if _, err := os.Stat(oldPrivVal); !os.IsNotExist(err) {
  84. oldPV, err := privval.LoadOldFilePV(oldPrivVal)
  85. if err != nil {
  86. return nil, fmt.Errorf("Error reading OldPrivValidator from %v: %v\n", oldPrivVal, err)
  87. }
  88. logger.Info("Upgrading PrivValidator file",
  89. "old", oldPrivVal,
  90. "newKey", newPrivValKey,
  91. "newState", newPrivValState,
  92. )
  93. oldPV.Upgrade(newPrivValKey, newPrivValState)
  94. }
  95. return NewNode(config,
  96. privval.LoadOrGenFilePV(newPrivValKey, newPrivValState),
  97. nodeKey,
  98. proxy.DefaultClientCreator(config.ProxyApp, config.ABCI, config.DBDir()),
  99. DefaultGenesisDocProviderFunc(config),
  100. DefaultDBProvider,
  101. DefaultMetricsProvider(config.Instrumentation),
  102. logger,
  103. )
  104. }
  105. // MetricsProvider returns a consensus, p2p and mempool Metrics.
  106. type MetricsProvider func() (*cs.Metrics, *p2p.Metrics, *mempl.Metrics, *sm.Metrics)
  107. // DefaultMetricsProvider returns Metrics build using Prometheus client library
  108. // if Prometheus is enabled. Otherwise, it returns no-op Metrics.
  109. func DefaultMetricsProvider(config *cfg.InstrumentationConfig) MetricsProvider {
  110. return func() (*cs.Metrics, *p2p.Metrics, *mempl.Metrics, *sm.Metrics) {
  111. if config.Prometheus {
  112. return cs.PrometheusMetrics(config.Namespace), p2p.PrometheusMetrics(config.Namespace),
  113. mempl.PrometheusMetrics(config.Namespace), sm.PrometheusMetrics(config.Namespace)
  114. }
  115. return cs.NopMetrics(), p2p.NopMetrics(), mempl.NopMetrics(), sm.NopMetrics()
  116. }
  117. }
  118. //------------------------------------------------------------------------------
  119. // Node is the highest level interface to a full Tendermint node.
  120. // It includes all configuration information and running services.
  121. type Node struct {
  122. cmn.BaseService
  123. // config
  124. config *cfg.Config
  125. genesisDoc *types.GenesisDoc // initial validator set
  126. privValidator types.PrivValidator // local node's validator key
  127. // network
  128. transport *p2p.MultiplexTransport
  129. sw *p2p.Switch // p2p connections
  130. addrBook pex.AddrBook // known peers
  131. nodeInfo p2p.NodeInfo
  132. nodeKey *p2p.NodeKey // our node privkey
  133. isListening bool
  134. // services
  135. eventBus *types.EventBus // pub/sub for services
  136. stateDB dbm.DB
  137. blockStore *bc.BlockStore // store the blockchain to disk
  138. bcReactor *bc.BlockchainReactor // for fast-syncing
  139. mempoolReactor *mempl.MempoolReactor // for gossipping transactions
  140. consensusState *cs.ConsensusState // latest consensus state
  141. consensusReactor *cs.ConsensusReactor // for participating in the consensus
  142. evidencePool *evidence.EvidencePool // tracking evidence
  143. proxyApp proxy.AppConns // connection to the application
  144. rpcListeners []net.Listener // rpc servers
  145. txIndexer txindex.TxIndexer
  146. indexerService *txindex.IndexerService
  147. prometheusSrv *http.Server
  148. }
  149. // NewNode returns a new, ready to go, Tendermint Node.
  150. func NewNode(config *cfg.Config,
  151. privValidator types.PrivValidator,
  152. nodeKey *p2p.NodeKey,
  153. clientCreator proxy.ClientCreator,
  154. genesisDocProvider GenesisDocProvider,
  155. dbProvider DBProvider,
  156. metricsProvider MetricsProvider,
  157. logger log.Logger) (*Node, error) {
  158. // Get BlockStore
  159. blockStoreDB, err := dbProvider(&DBContext{"blockstore", config})
  160. if err != nil {
  161. return nil, err
  162. }
  163. blockStore := bc.NewBlockStore(blockStoreDB)
  164. // Get State
  165. stateDB, err := dbProvider(&DBContext{"state", config})
  166. if err != nil {
  167. return nil, err
  168. }
  169. // Get genesis doc
  170. // TODO: move to state package?
  171. genDoc, err := loadGenesisDoc(stateDB)
  172. if err != nil {
  173. genDoc, err = genesisDocProvider()
  174. if err != nil {
  175. return nil, err
  176. }
  177. // save genesis doc to prevent a certain class of user errors (e.g. when it
  178. // was changed, accidentally or not). Also good for audit trail.
  179. saveGenesisDoc(stateDB, genDoc)
  180. }
  181. state, err := sm.LoadStateFromDBOrGenesisDoc(stateDB, genDoc)
  182. if err != nil {
  183. return nil, err
  184. }
  185. // Create the proxyApp and establish connections to the ABCI app (consensus, mempool, query).
  186. proxyApp := proxy.NewAppConns(clientCreator)
  187. proxyApp.SetLogger(logger.With("module", "proxy"))
  188. if err := proxyApp.Start(); err != nil {
  189. return nil, fmt.Errorf("Error starting proxy app connections: %v", err)
  190. }
  191. // Create the handshaker, which calls RequestInfo, sets the AppVersion on the state,
  192. // and replays any blocks as necessary to sync tendermint with the app.
  193. consensusLogger := logger.With("module", "consensus")
  194. handshaker := cs.NewHandshaker(stateDB, state, blockStore, genDoc)
  195. handshaker.SetLogger(consensusLogger)
  196. if err := handshaker.Handshake(proxyApp); err != nil {
  197. return nil, fmt.Errorf("Error during handshake: %v", err)
  198. }
  199. // Reload the state. It will have the Version.Consensus.App set by the
  200. // Handshake, and may have other modifications as well (ie. depending on
  201. // what happened during block replay).
  202. state = sm.LoadState(stateDB)
  203. // Log the version info.
  204. logger.Info("Version info",
  205. "software", version.TMCoreSemVer,
  206. "block", version.BlockProtocol,
  207. "p2p", version.P2PProtocol,
  208. )
  209. // If the state and software differ in block version, at least log it.
  210. if state.Version.Consensus.Block != version.BlockProtocol {
  211. logger.Info("Software and state have different block protocols",
  212. "software", version.BlockProtocol,
  213. "state", state.Version.Consensus.Block,
  214. )
  215. }
  216. if config.PrivValidatorListenAddr != "" {
  217. // If an address is provided, listen on the socket for a connection from an
  218. // external signing process.
  219. // FIXME: we should start services inside OnStart
  220. privValidator, err = createAndStartPrivValidatorSocketClient(config.PrivValidatorListenAddr, logger)
  221. if err != nil {
  222. return nil, errors.Wrap(err, "Error with private validator socket client")
  223. }
  224. }
  225. // Decide whether to fast-sync or not
  226. // We don't fast-sync when the only validator is us.
  227. fastSync := config.FastSync
  228. if state.Validators.Size() == 1 {
  229. addr, _ := state.Validators.GetByIndex(0)
  230. privValAddr := privValidator.GetPubKey().Address()
  231. if bytes.Equal(privValAddr, addr) {
  232. fastSync = false
  233. }
  234. }
  235. pubKey := privValidator.GetPubKey()
  236. addr := pubKey.Address()
  237. // Log whether this node is a validator or an observer
  238. if state.Validators.HasAddress(addr) {
  239. consensusLogger.Info("This node is a validator", "addr", addr, "pubKey", pubKey)
  240. } else {
  241. consensusLogger.Info("This node is not a validator", "addr", addr, "pubKey", pubKey)
  242. }
  243. csMetrics, p2pMetrics, memplMetrics, smMetrics := metricsProvider()
  244. // Make MempoolReactor
  245. mempool := mempl.NewMempool(
  246. config.Mempool,
  247. proxyApp.Mempool(),
  248. state.LastBlockHeight,
  249. mempl.WithMetrics(memplMetrics),
  250. mempl.WithPreCheck(sm.TxPreCheck(state)),
  251. mempl.WithPostCheck(sm.TxPostCheck(state)),
  252. )
  253. mempoolLogger := logger.With("module", "mempool")
  254. mempool.SetLogger(mempoolLogger)
  255. if config.Mempool.WalEnabled() {
  256. mempool.InitWAL() // no need to have the mempool wal during tests
  257. }
  258. mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool)
  259. mempoolReactor.SetLogger(mempoolLogger)
  260. if config.Consensus.WaitForTxs() {
  261. mempool.EnableTxsAvailable()
  262. }
  263. // Make Evidence Reactor
  264. evidenceDB, err := dbProvider(&DBContext{"evidence", config})
  265. if err != nil {
  266. return nil, err
  267. }
  268. evidenceLogger := logger.With("module", "evidence")
  269. evidenceStore := evidence.NewEvidenceStore(evidenceDB)
  270. evidencePool := evidence.NewEvidencePool(stateDB, evidenceStore)
  271. evidencePool.SetLogger(evidenceLogger)
  272. evidenceReactor := evidence.NewEvidenceReactor(evidencePool)
  273. evidenceReactor.SetLogger(evidenceLogger)
  274. blockExecLogger := logger.With("module", "state")
  275. // make block executor for consensus and blockchain reactors to execute blocks
  276. blockExec := sm.NewBlockExecutor(
  277. stateDB,
  278. blockExecLogger,
  279. proxyApp.Consensus(),
  280. mempool,
  281. evidencePool,
  282. sm.BlockExecutorWithMetrics(smMetrics),
  283. )
  284. // Make BlockchainReactor
  285. bcReactor := bc.NewBlockchainReactor(state.Copy(), blockExec, blockStore, fastSync)
  286. bcReactor.SetLogger(logger.With("module", "blockchain"))
  287. // Make ConsensusReactor
  288. consensusState := cs.NewConsensusState(
  289. config.Consensus,
  290. state.Copy(),
  291. blockExec,
  292. blockStore,
  293. mempool,
  294. evidencePool,
  295. cs.StateMetrics(csMetrics),
  296. )
  297. consensusState.SetLogger(consensusLogger)
  298. if privValidator != nil {
  299. consensusState.SetPrivValidator(privValidator)
  300. }
  301. consensusReactor := cs.NewConsensusReactor(consensusState, fastSync, cs.ReactorMetrics(csMetrics))
  302. consensusReactor.SetLogger(consensusLogger)
  303. eventBus := types.NewEventBus()
  304. eventBus.SetLogger(logger.With("module", "events"))
  305. // services which will be publishing and/or subscribing for messages (events)
  306. // consensusReactor will set it on consensusState and blockExecutor
  307. consensusReactor.SetEventBus(eventBus)
  308. // Transaction indexing
  309. var txIndexer txindex.TxIndexer
  310. switch config.TxIndex.Indexer {
  311. case "kv":
  312. store, err := dbProvider(&DBContext{"tx_index", config})
  313. if err != nil {
  314. return nil, err
  315. }
  316. if config.TxIndex.IndexTags != "" {
  317. txIndexer = kv.NewTxIndex(store, kv.IndexTags(splitAndTrimEmpty(config.TxIndex.IndexTags, ",", " ")))
  318. } else if config.TxIndex.IndexAllTags {
  319. txIndexer = kv.NewTxIndex(store, kv.IndexAllTags())
  320. } else {
  321. txIndexer = kv.NewTxIndex(store)
  322. }
  323. default:
  324. txIndexer = &null.TxIndex{}
  325. }
  326. indexerService := txindex.NewIndexerService(txIndexer, eventBus)
  327. indexerService.SetLogger(logger.With("module", "txindex"))
  328. p2pLogger := logger.With("module", "p2p")
  329. nodeInfo, err := makeNodeInfo(
  330. config,
  331. nodeKey.ID(),
  332. txIndexer,
  333. genDoc.ChainID,
  334. p2p.NewProtocolVersion(
  335. version.P2PProtocol, // global
  336. state.Version.Consensus.Block,
  337. state.Version.Consensus.App,
  338. ),
  339. )
  340. if err != nil {
  341. return nil, err
  342. }
  343. // Setup Transport.
  344. var (
  345. mConnConfig = p2p.MConnConfig(config.P2P)
  346. transport = p2p.NewMultiplexTransport(nodeInfo, *nodeKey, mConnConfig)
  347. connFilters = []p2p.ConnFilterFunc{}
  348. peerFilters = []p2p.PeerFilterFunc{}
  349. )
  350. if !config.P2P.AllowDuplicateIP {
  351. connFilters = append(connFilters, p2p.ConnDuplicateIPFilter())
  352. }
  353. // Filter peers by addr or pubkey with an ABCI query.
  354. // If the query return code is OK, add peer.
  355. if config.FilterPeers {
  356. connFilters = append(
  357. connFilters,
  358. // ABCI query for address filtering.
  359. func(_ p2p.ConnSet, c net.Conn, _ []net.IP) error {
  360. res, err := proxyApp.Query().QuerySync(abci.RequestQuery{
  361. Path: fmt.Sprintf("/p2p/filter/addr/%s", c.RemoteAddr().String()),
  362. })
  363. if err != nil {
  364. return err
  365. }
  366. if res.IsErr() {
  367. return fmt.Errorf("Error querying abci app: %v", res)
  368. }
  369. return nil
  370. },
  371. )
  372. peerFilters = append(
  373. peerFilters,
  374. // ABCI query for ID filtering.
  375. func(_ p2p.IPeerSet, p p2p.Peer) error {
  376. res, err := proxyApp.Query().QuerySync(abci.RequestQuery{
  377. Path: fmt.Sprintf("/p2p/filter/id/%s", p.ID()),
  378. })
  379. if err != nil {
  380. return err
  381. }
  382. if res.IsErr() {
  383. return fmt.Errorf("Error querying abci app: %v", res)
  384. }
  385. return nil
  386. },
  387. )
  388. }
  389. p2p.MultiplexTransportConnFilters(connFilters...)(transport)
  390. // Setup Switch.
  391. sw := p2p.NewSwitch(
  392. config.P2P,
  393. transport,
  394. p2p.WithMetrics(p2pMetrics),
  395. p2p.SwitchPeerFilters(peerFilters...),
  396. )
  397. sw.SetLogger(p2pLogger)
  398. sw.AddReactor("MEMPOOL", mempoolReactor)
  399. sw.AddReactor("BLOCKCHAIN", bcReactor)
  400. sw.AddReactor("CONSENSUS", consensusReactor)
  401. sw.AddReactor("EVIDENCE", evidenceReactor)
  402. sw.SetNodeInfo(nodeInfo)
  403. sw.SetNodeKey(nodeKey)
  404. p2pLogger.Info("P2P Node ID", "ID", nodeKey.ID(), "file", config.NodeKeyFile())
  405. // Optionally, start the pex reactor
  406. //
  407. // TODO:
  408. //
  409. // We need to set Seeds and PersistentPeers on the switch,
  410. // since it needs to be able to use these (and their DNS names)
  411. // even if the PEX is off. We can include the DNS name in the NetAddress,
  412. // but it would still be nice to have a clear list of the current "PersistentPeers"
  413. // somewhere that we can return with net_info.
  414. //
  415. // If PEX is on, it should handle dialing the seeds. Otherwise the switch does it.
  416. // Note we currently use the addrBook regardless at least for AddOurAddress
  417. addrBook := pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
  418. // Add ourselves to addrbook to prevent dialing ourselves
  419. addrBook.AddOurAddress(nodeInfo.NetAddress())
  420. addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
  421. if config.P2P.PexReactor {
  422. // TODO persistent peers ? so we can have their DNS addrs saved
  423. pexReactor := pex.NewPEXReactor(addrBook,
  424. &pex.PEXReactorConfig{
  425. Seeds: splitAndTrimEmpty(config.P2P.Seeds, ",", " "),
  426. SeedMode: config.P2P.SeedMode,
  427. })
  428. pexReactor.SetLogger(logger.With("module", "pex"))
  429. sw.AddReactor("PEX", pexReactor)
  430. }
  431. sw.SetAddrBook(addrBook)
  432. // run the profile server
  433. profileHost := config.ProfListenAddress
  434. if profileHost != "" {
  435. go func() {
  436. logger.Error("Profile server", "err", http.ListenAndServe(profileHost, nil))
  437. }()
  438. }
  439. node := &Node{
  440. config: config,
  441. genesisDoc: genDoc,
  442. privValidator: privValidator,
  443. transport: transport,
  444. sw: sw,
  445. addrBook: addrBook,
  446. nodeInfo: nodeInfo,
  447. nodeKey: nodeKey,
  448. stateDB: stateDB,
  449. blockStore: blockStore,
  450. bcReactor: bcReactor,
  451. mempoolReactor: mempoolReactor,
  452. consensusState: consensusState,
  453. consensusReactor: consensusReactor,
  454. evidencePool: evidencePool,
  455. proxyApp: proxyApp,
  456. txIndexer: txIndexer,
  457. indexerService: indexerService,
  458. eventBus: eventBus,
  459. }
  460. node.BaseService = *cmn.NewBaseService(logger, "Node", node)
  461. return node, nil
  462. }
  463. // OnStart starts the Node. It implements cmn.Service.
  464. func (n *Node) OnStart() error {
  465. now := tmtime.Now()
  466. genTime := n.genesisDoc.GenesisTime
  467. if genTime.After(now) {
  468. n.Logger.Info("Genesis time is in the future. Sleeping until then...", "genTime", genTime)
  469. time.Sleep(genTime.Sub(now))
  470. }
  471. err := n.eventBus.Start()
  472. if err != nil {
  473. return err
  474. }
  475. // Add private IDs to addrbook to block those peers being added
  476. n.addrBook.AddPrivateIDs(splitAndTrimEmpty(n.config.P2P.PrivatePeerIDs, ",", " "))
  477. // Start the RPC server before the P2P server
  478. // so we can eg. receive txs for the first block
  479. if n.config.RPC.ListenAddress != "" {
  480. listeners, err := n.startRPC()
  481. if err != nil {
  482. return err
  483. }
  484. n.rpcListeners = listeners
  485. }
  486. if n.config.Instrumentation.Prometheus &&
  487. n.config.Instrumentation.PrometheusListenAddr != "" {
  488. n.prometheusSrv = n.startPrometheusServer(n.config.Instrumentation.PrometheusListenAddr)
  489. }
  490. // Start the transport.
  491. addr, err := p2p.NewNetAddressStringWithOptionalID(n.config.P2P.ListenAddress)
  492. if err != nil {
  493. return err
  494. }
  495. if err := n.transport.Listen(*addr); err != nil {
  496. return err
  497. }
  498. n.isListening = true
  499. // Start the switch (the P2P server).
  500. err = n.sw.Start()
  501. if err != nil {
  502. return err
  503. }
  504. // Always connect to persistent peers
  505. if n.config.P2P.PersistentPeers != "" {
  506. err = n.sw.DialPeersAsync(n.addrBook, splitAndTrimEmpty(n.config.P2P.PersistentPeers, ",", " "), true)
  507. if err != nil {
  508. return err
  509. }
  510. }
  511. // start tx indexer
  512. return n.indexerService.Start()
  513. }
  514. // OnStop stops the Node. It implements cmn.Service.
  515. func (n *Node) OnStop() {
  516. n.BaseService.OnStop()
  517. n.Logger.Info("Stopping Node")
  518. // first stop the non-reactor services
  519. n.eventBus.Stop()
  520. n.indexerService.Stop()
  521. // now stop the reactors
  522. // TODO: gracefully disconnect from peers.
  523. n.sw.Stop()
  524. // stop mempool WAL
  525. if n.config.Mempool.WalEnabled() {
  526. n.mempoolReactor.Mempool.CloseWAL()
  527. }
  528. if err := n.transport.Close(); err != nil {
  529. n.Logger.Error("Error closing transport", "err", err)
  530. }
  531. n.isListening = false
  532. // finally stop the listeners / external services
  533. for _, l := range n.rpcListeners {
  534. n.Logger.Info("Closing rpc listener", "listener", l)
  535. if err := l.Close(); err != nil {
  536. n.Logger.Error("Error closing listener", "listener", l, "err", err)
  537. }
  538. }
  539. if pvsc, ok := n.privValidator.(cmn.Service); ok {
  540. pvsc.Stop()
  541. }
  542. if n.prometheusSrv != nil {
  543. if err := n.prometheusSrv.Shutdown(context.Background()); err != nil {
  544. // Error from closing listeners, or context timeout:
  545. n.Logger.Error("Prometheus HTTP server Shutdown", "err", err)
  546. }
  547. }
  548. }
  549. // ConfigureRPC sets all variables in rpccore so they will serve
  550. // rpc calls from this node
  551. func (n *Node) ConfigureRPC() {
  552. rpccore.SetStateDB(n.stateDB)
  553. rpccore.SetBlockStore(n.blockStore)
  554. rpccore.SetConsensusState(n.consensusState)
  555. rpccore.SetMempool(n.mempoolReactor.Mempool)
  556. rpccore.SetEvidencePool(n.evidencePool)
  557. rpccore.SetP2PPeers(n.sw)
  558. rpccore.SetP2PTransport(n)
  559. pubKey := n.privValidator.GetPubKey()
  560. rpccore.SetPubKey(pubKey)
  561. rpccore.SetGenesisDoc(n.genesisDoc)
  562. rpccore.SetAddrBook(n.addrBook)
  563. rpccore.SetProxyAppQuery(n.proxyApp.Query())
  564. rpccore.SetTxIndexer(n.txIndexer)
  565. rpccore.SetConsensusReactor(n.consensusReactor)
  566. rpccore.SetEventBus(n.eventBus)
  567. rpccore.SetLogger(n.Logger.With("module", "rpc"))
  568. }
  569. func (n *Node) startRPC() ([]net.Listener, error) {
  570. n.ConfigureRPC()
  571. listenAddrs := splitAndTrimEmpty(n.config.RPC.ListenAddress, ",", " ")
  572. coreCodec := amino.NewCodec()
  573. ctypes.RegisterAmino(coreCodec)
  574. if n.config.RPC.Unsafe {
  575. rpccore.AddUnsafeRoutes()
  576. }
  577. // we may expose the rpc over both a unix and tcp socket
  578. listeners := make([]net.Listener, len(listenAddrs))
  579. for i, listenAddr := range listenAddrs {
  580. mux := http.NewServeMux()
  581. rpcLogger := n.Logger.With("module", "rpc-server")
  582. wm := rpcserver.NewWebsocketManager(rpccore.Routes, coreCodec, rpcserver.EventSubscriber(n.eventBus))
  583. wm.SetLogger(rpcLogger.With("protocol", "websocket"))
  584. mux.HandleFunc("/websocket", wm.WebsocketHandler)
  585. rpcserver.RegisterRPCFuncs(mux, rpccore.Routes, coreCodec, rpcLogger)
  586. listener, err := rpcserver.Listen(
  587. listenAddr,
  588. rpcserver.Config{MaxOpenConnections: n.config.RPC.MaxOpenConnections},
  589. )
  590. if err != nil {
  591. return nil, err
  592. }
  593. var rootHandler http.Handler = mux
  594. if n.config.RPC.IsCorsEnabled() {
  595. corsMiddleware := cors.New(cors.Options{
  596. AllowedOrigins: n.config.RPC.CORSAllowedOrigins,
  597. AllowedMethods: n.config.RPC.CORSAllowedMethods,
  598. AllowedHeaders: n.config.RPC.CORSAllowedHeaders,
  599. })
  600. rootHandler = corsMiddleware.Handler(mux)
  601. }
  602. go rpcserver.StartHTTPServer(
  603. listener,
  604. rootHandler,
  605. rpcLogger,
  606. )
  607. listeners[i] = listener
  608. }
  609. // we expose a simplified api over grpc for convenience to app devs
  610. grpcListenAddr := n.config.RPC.GRPCListenAddress
  611. if grpcListenAddr != "" {
  612. listener, err := rpcserver.Listen(
  613. grpcListenAddr, rpcserver.Config{MaxOpenConnections: n.config.RPC.GRPCMaxOpenConnections})
  614. if err != nil {
  615. return nil, err
  616. }
  617. go grpccore.StartGRPCServer(listener)
  618. listeners = append(listeners, listener)
  619. }
  620. return listeners, nil
  621. }
  622. // startPrometheusServer starts a Prometheus HTTP server, listening for metrics
  623. // collectors on addr.
  624. func (n *Node) startPrometheusServer(addr string) *http.Server {
  625. srv := &http.Server{
  626. Addr: addr,
  627. Handler: promhttp.InstrumentMetricHandler(
  628. prometheus.DefaultRegisterer, promhttp.HandlerFor(
  629. prometheus.DefaultGatherer,
  630. promhttp.HandlerOpts{MaxRequestsInFlight: n.config.Instrumentation.MaxOpenConnections},
  631. ),
  632. ),
  633. }
  634. go func() {
  635. if err := srv.ListenAndServe(); err != http.ErrServerClosed {
  636. // Error starting or closing listener:
  637. n.Logger.Error("Prometheus HTTP server ListenAndServe", "err", err)
  638. }
  639. }()
  640. return srv
  641. }
  642. // Switch returns the Node's Switch.
  643. func (n *Node) Switch() *p2p.Switch {
  644. return n.sw
  645. }
  646. // BlockStore returns the Node's BlockStore.
  647. func (n *Node) BlockStore() *bc.BlockStore {
  648. return n.blockStore
  649. }
  650. // ConsensusState returns the Node's ConsensusState.
  651. func (n *Node) ConsensusState() *cs.ConsensusState {
  652. return n.consensusState
  653. }
  654. // ConsensusReactor returns the Node's ConsensusReactor.
  655. func (n *Node) ConsensusReactor() *cs.ConsensusReactor {
  656. return n.consensusReactor
  657. }
  658. // MempoolReactor returns the Node's MempoolReactor.
  659. func (n *Node) MempoolReactor() *mempl.MempoolReactor {
  660. return n.mempoolReactor
  661. }
  662. // EvidencePool returns the Node's EvidencePool.
  663. func (n *Node) EvidencePool() *evidence.EvidencePool {
  664. return n.evidencePool
  665. }
  666. // EventBus returns the Node's EventBus.
  667. func (n *Node) EventBus() *types.EventBus {
  668. return n.eventBus
  669. }
  670. // PrivValidator returns the Node's PrivValidator.
  671. // XXX: for convenience only!
  672. func (n *Node) PrivValidator() types.PrivValidator {
  673. return n.privValidator
  674. }
  675. // GenesisDoc returns the Node's GenesisDoc.
  676. func (n *Node) GenesisDoc() *types.GenesisDoc {
  677. return n.genesisDoc
  678. }
  679. // ProxyApp returns the Node's AppConns, representing its connections to the ABCI application.
  680. func (n *Node) ProxyApp() proxy.AppConns {
  681. return n.proxyApp
  682. }
  683. //------------------------------------------------------------------------------
  684. func (n *Node) Listeners() []string {
  685. return []string{
  686. fmt.Sprintf("Listener(@%v)", n.config.P2P.ExternalAddress),
  687. }
  688. }
  689. func (n *Node) IsListening() bool {
  690. return n.isListening
  691. }
  692. // NodeInfo returns the Node's Info from the Switch.
  693. func (n *Node) NodeInfo() p2p.NodeInfo {
  694. return n.nodeInfo
  695. }
  696. func makeNodeInfo(
  697. config *cfg.Config,
  698. nodeID p2p.ID,
  699. txIndexer txindex.TxIndexer,
  700. chainID string,
  701. protocolVersion p2p.ProtocolVersion,
  702. ) (p2p.NodeInfo, error) {
  703. txIndexerStatus := "on"
  704. if _, ok := txIndexer.(*null.TxIndex); ok {
  705. txIndexerStatus = "off"
  706. }
  707. nodeInfo := p2p.DefaultNodeInfo{
  708. ProtocolVersion: protocolVersion,
  709. ID_: nodeID,
  710. Network: chainID,
  711. Version: version.TMCoreSemVer,
  712. Channels: []byte{
  713. bc.BlockchainChannel,
  714. cs.StateChannel, cs.DataChannel, cs.VoteChannel, cs.VoteSetBitsChannel,
  715. mempl.MempoolChannel,
  716. evidence.EvidenceChannel,
  717. },
  718. Moniker: config.Moniker,
  719. Other: p2p.DefaultNodeInfoOther{
  720. TxIndex: txIndexerStatus,
  721. RPCAddress: config.RPC.ListenAddress,
  722. },
  723. }
  724. if config.P2P.PexReactor {
  725. nodeInfo.Channels = append(nodeInfo.Channels, pex.PexChannel)
  726. }
  727. lAddr := config.P2P.ExternalAddress
  728. if lAddr == "" {
  729. lAddr = config.P2P.ListenAddress
  730. }
  731. nodeInfo.ListenAddr = lAddr
  732. err := nodeInfo.Validate()
  733. return nodeInfo, err
  734. }
  735. //------------------------------------------------------------------------------
  736. var (
  737. genesisDocKey = []byte("genesisDoc")
  738. )
  739. // panics if failed to unmarshal bytes
  740. func loadGenesisDoc(db dbm.DB) (*types.GenesisDoc, error) {
  741. bytes := db.Get(genesisDocKey)
  742. if len(bytes) == 0 {
  743. return nil, errors.New("Genesis doc not found")
  744. }
  745. var genDoc *types.GenesisDoc
  746. err := cdc.UnmarshalJSON(bytes, &genDoc)
  747. if err != nil {
  748. cmn.PanicCrisis(fmt.Sprintf("Failed to load genesis doc due to unmarshaling error: %v (bytes: %X)", err, bytes))
  749. }
  750. return genDoc, nil
  751. }
  752. // panics if failed to marshal the given genesis document
  753. func saveGenesisDoc(db dbm.DB, genDoc *types.GenesisDoc) {
  754. bytes, err := cdc.MarshalJSON(genDoc)
  755. if err != nil {
  756. cmn.PanicCrisis(fmt.Sprintf("Failed to save genesis doc due to marshaling error: %v", err))
  757. }
  758. db.SetSync(genesisDocKey, bytes)
  759. }
  760. func createAndStartPrivValidatorSocketClient(
  761. listenAddr string,
  762. logger log.Logger,
  763. ) (types.PrivValidator, error) {
  764. var listener net.Listener
  765. protocol, address := cmn.ProtocolAndAddress(listenAddr)
  766. ln, err := net.Listen(protocol, address)
  767. if err != nil {
  768. return nil, err
  769. }
  770. switch protocol {
  771. case "unix":
  772. listener = privval.NewUnixListener(ln)
  773. case "tcp":
  774. // TODO: persist this key so external signer
  775. // can actually authenticate us
  776. listener = privval.NewTCPListener(ln, ed25519.GenPrivKey())
  777. default:
  778. return nil, fmt.Errorf(
  779. "Wrong listen address: expected either 'tcp' or 'unix' protocols, got %s",
  780. protocol,
  781. )
  782. }
  783. pvsc := privval.NewSocketVal(logger.With("module", "privval"), listener)
  784. if err := pvsc.Start(); err != nil {
  785. return nil, errors.Wrap(err, "failed to start private validator")
  786. }
  787. return pvsc, nil
  788. }
  789. // splitAndTrimEmpty slices s into all subslices separated by sep and returns a
  790. // slice of the string s with all leading and trailing Unicode code points
  791. // contained in cutset removed. If sep is empty, SplitAndTrim splits after each
  792. // UTF-8 sequence. First part is equivalent to strings.SplitN with a count of
  793. // -1. also filter out empty strings, only return non-empty strings.
  794. func splitAndTrimEmpty(s, sep, cutset string) []string {
  795. if s == "" {
  796. return []string{}
  797. }
  798. spl := strings.Split(s, sep)
  799. nonEmptyStrings := make([]string, 0, len(spl))
  800. for i := 0; i < len(spl); i++ {
  801. element := strings.Trim(spl[i], cutset)
  802. if element != "" {
  803. nonEmptyStrings = append(nonEmptyStrings, element)
  804. }
  805. }
  806. return nonEmptyStrings
  807. }