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.

413 lines
14 KiB

limit number of /subscribe clients and queries per client (#3269) * limit number of /subscribe clients and queries per client Add the following config variables (under [rpc] section): * max_subscription_clients * max_subscriptions_per_client * timeout_broadcast_tx_commit Fixes #2826 new HTTPClient interface for subscriptions finalize HTTPClient events interface remove EventSubscriber fix data race ``` WARNING: DATA RACE Read at 0x00c000a36060 by goroutine 129: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe.func1() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:168 +0x1f0 Previous write at 0x00c000a36060 by goroutine 132: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:191 +0x4e0 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 129 (running) created at: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:164 +0x4b7 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 132 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:119 +0x186 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ================== ``` lite client works (tested manually) godoc comments httpclient: do not close the out channel use TimeoutBroadcastTxCommit no timeout for unsubscribe but 1s Local (5s HTTP) timeout for resubscribe format code change Subscribe#out cap to 1 and replace config vars with RPCConfig TimeoutBroadcastTxCommit can't be greater than rpcserver.WriteTimeout rpc: Context as first parameter to all functions reformat code fixes after my own review fixes after Ethan's review add test stubs fix config.toml * fixes after manual testing - rpc: do not recommend to use BroadcastTxCommit because it's slow and wastes Tendermint resources (pubsub) - rpc: better error in Subscribe and BroadcastTxCommit - HTTPClient: do not resubscribe if err = ErrAlreadySubscribed * fixes after Ismail's review * Update rpc/grpc/grpc_test.go Co-Authored-By: melekes <anton.kalyaev@gmail.com>
6 years ago
  1. package config
  2. import (
  3. "bytes"
  4. "fmt"
  5. "io/ioutil"
  6. "path/filepath"
  7. "text/template"
  8. cmn "github.com/tendermint/tendermint/libs/common"
  9. )
  10. // DefaultDirPerm is the default permissions used when creating directories.
  11. const DefaultDirPerm = 0700
  12. var configTemplate *template.Template
  13. func init() {
  14. var err error
  15. if configTemplate, err = template.New("configFileTemplate").Parse(defaultConfigTemplate); err != nil {
  16. panic(err)
  17. }
  18. }
  19. /****** these are for production settings ***********/
  20. // EnsureRoot creates the root, config, and data directories if they don't exist,
  21. // and panics if it fails.
  22. func EnsureRoot(rootDir string) {
  23. if err := cmn.EnsureDir(rootDir, DefaultDirPerm); err != nil {
  24. cmn.PanicSanity(err.Error())
  25. }
  26. if err := cmn.EnsureDir(filepath.Join(rootDir, defaultConfigDir), DefaultDirPerm); err != nil {
  27. cmn.PanicSanity(err.Error())
  28. }
  29. if err := cmn.EnsureDir(filepath.Join(rootDir, defaultDataDir), DefaultDirPerm); err != nil {
  30. cmn.PanicSanity(err.Error())
  31. }
  32. configFilePath := filepath.Join(rootDir, defaultConfigFilePath)
  33. // Write default config file if missing.
  34. if !cmn.FileExists(configFilePath) {
  35. writeDefaultConfigFile(configFilePath)
  36. }
  37. }
  38. // XXX: this func should probably be called by cmd/tendermint/commands/init.go
  39. // alongside the writing of the genesis.json and priv_validator.json
  40. func writeDefaultConfigFile(configFilePath string) {
  41. WriteConfigFile(configFilePath, DefaultConfig())
  42. }
  43. // WriteConfigFile renders config using the template and writes it to configFilePath.
  44. func WriteConfigFile(configFilePath string, config *Config) {
  45. var buffer bytes.Buffer
  46. if err := configTemplate.Execute(&buffer, config); err != nil {
  47. panic(err)
  48. }
  49. cmn.MustWriteFile(configFilePath, buffer.Bytes(), 0644)
  50. }
  51. // Note: any changes to the comments/variables/mapstructure
  52. // must be reflected in the appropriate struct in config/config.go
  53. const defaultConfigTemplate = `# This is a TOML config file.
  54. # For more information, see https://github.com/toml-lang/toml
  55. ##### main base config options #####
  56. # TCP or UNIX socket address of the ABCI application,
  57. # or the name of an ABCI application compiled in with the Tendermint binary
  58. proxy_app = "{{ .BaseConfig.ProxyApp }}"
  59. # A custom human readable name for this node
  60. moniker = "{{ .BaseConfig.Moniker }}"
  61. # If this node is many blocks behind the tip of the chain, FastSync
  62. # allows them to catchup quickly by downloading blocks in parallel
  63. # and verifying their commits
  64. fast_sync = {{ .BaseConfig.FastSync }}
  65. # Database backend: leveldb | memdb | cleveldb
  66. db_backend = "{{ .BaseConfig.DBBackend }}"
  67. # Database directory
  68. db_dir = "{{ js .BaseConfig.DBPath }}"
  69. # Output level for logging, including package level options
  70. log_level = "{{ .BaseConfig.LogLevel }}"
  71. # Output format: 'plain' (colored text) or 'json'
  72. log_format = "{{ .BaseConfig.LogFormat }}"
  73. ##### additional base config options #####
  74. # Path to the JSON file containing the initial validator set and other meta data
  75. genesis_file = "{{ js .BaseConfig.Genesis }}"
  76. # Path to the JSON file containing the private key to use as a validator in the consensus protocol
  77. priv_validator_key_file = "{{ js .BaseConfig.PrivValidatorKey }}"
  78. # Path to the JSON file containing the last sign state of a validator
  79. priv_validator_state_file = "{{ js .BaseConfig.PrivValidatorState }}"
  80. # TCP or UNIX socket address for Tendermint to listen on for
  81. # connections from an external PrivValidator process
  82. priv_validator_laddr = "{{ .BaseConfig.PrivValidatorListenAddr }}"
  83. # Path to the JSON file containing the private key to use for node authentication in the p2p protocol
  84. node_key_file = "{{ js .BaseConfig.NodeKey }}"
  85. # Mechanism to connect to the ABCI application: socket | grpc
  86. abci = "{{ .BaseConfig.ABCI }}"
  87. # TCP or UNIX socket address for the profiling server to listen on
  88. prof_laddr = "{{ .BaseConfig.ProfListenAddress }}"
  89. # If true, query the ABCI app on connecting to a new peer
  90. # so the app can decide if we should keep the connection or not
  91. filter_peers = {{ .BaseConfig.FilterPeers }}
  92. ##### advanced configuration options #####
  93. ##### rpc server configuration options #####
  94. [rpc]
  95. # TCP or UNIX socket address for the RPC server to listen on
  96. laddr = "{{ .RPC.ListenAddress }}"
  97. # A list of origins a cross-domain request can be executed from
  98. # Default value '[]' disables cors support
  99. # Use '["*"]' to allow any origin
  100. cors_allowed_origins = [{{ range .RPC.CORSAllowedOrigins }}{{ printf "%q, " . }}{{end}}]
  101. # A list of methods the client is allowed to use with cross-domain requests
  102. cors_allowed_methods = [{{ range .RPC.CORSAllowedMethods }}{{ printf "%q, " . }}{{end}}]
  103. # A list of non simple headers the client is allowed to use with cross-domain requests
  104. cors_allowed_headers = [{{ range .RPC.CORSAllowedHeaders }}{{ printf "%q, " . }}{{end}}]
  105. # TCP or UNIX socket address for the gRPC server to listen on
  106. # NOTE: This server only supports /broadcast_tx_commit
  107. grpc_laddr = "{{ .RPC.GRPCListenAddress }}"
  108. # Maximum number of simultaneous connections.
  109. # Does not include RPC (HTTP&WebSocket) connections. See max_open_connections
  110. # If you want to accept a larger number than the default, make sure
  111. # you increase your OS limits.
  112. # 0 - unlimited.
  113. # Should be < {ulimit -Sn} - {MaxNumInboundPeers} - {MaxNumOutboundPeers} - {N of wal, db and other open files}
  114. # 1024 - 40 - 10 - 50 = 924 = ~900
  115. grpc_max_open_connections = {{ .RPC.GRPCMaxOpenConnections }}
  116. # Activate unsafe RPC commands like /dial_seeds and /unsafe_flush_mempool
  117. unsafe = {{ .RPC.Unsafe }}
  118. # Maximum number of simultaneous connections (including WebSocket).
  119. # Does not include gRPC connections. See grpc_max_open_connections
  120. # If you want to accept a larger number than the default, make sure
  121. # you increase your OS limits.
  122. # 0 - unlimited.
  123. # Should be < {ulimit -Sn} - {MaxNumInboundPeers} - {MaxNumOutboundPeers} - {N of wal, db and other open files}
  124. # 1024 - 40 - 10 - 50 = 924 = ~900
  125. max_open_connections = {{ .RPC.MaxOpenConnections }}
  126. # Maximum number of unique clientIDs that can /subscribe
  127. # If you're using /broadcast_tx_commit, set to the estimated maximum number
  128. # of broadcast_tx_commit calls per block.
  129. max_subscription_clients = {{ .RPC.MaxSubscriptionClients }}
  130. # Maximum number of unique queries a given client can /subscribe to
  131. # If you're using GRPC (or Local RPC client) and /broadcast_tx_commit, set to
  132. # the estimated # maximum number of broadcast_tx_commit calls per block.
  133. max_subscriptions_per_client = {{ .RPC.MaxSubscriptionsPerClient }}
  134. # How long to wait for a tx to be committed during /broadcast_tx_commit.
  135. timeout_broadcast_tx_commit = "{{ .RPC.TimeoutBroadcastTxCommit }}"
  136. ##### peer to peer configuration options #####
  137. [p2p]
  138. # Address to listen for incoming connections
  139. laddr = "{{ .P2P.ListenAddress }}"
  140. # Address to advertise to peers for them to dial
  141. # If empty, will use the same port as the laddr,
  142. # and will introspect on the listener or use UPnP
  143. # to figure out the address.
  144. external_address = "{{ .P2P.ExternalAddress }}"
  145. # Comma separated list of seed nodes to connect to
  146. seeds = "{{ .P2P.Seeds }}"
  147. # Comma separated list of nodes to keep persistent connections to
  148. persistent_peers = "{{ .P2P.PersistentPeers }}"
  149. # UPNP port forwarding
  150. upnp = {{ .P2P.UPNP }}
  151. # Path to address book
  152. addr_book_file = "{{ js .P2P.AddrBook }}"
  153. # Set true for strict address routability rules
  154. # Set false for private or local networks
  155. addr_book_strict = {{ .P2P.AddrBookStrict }}
  156. # Maximum number of inbound peers
  157. max_num_inbound_peers = {{ .P2P.MaxNumInboundPeers }}
  158. # Maximum number of outbound peers to connect to, excluding persistent peers
  159. max_num_outbound_peers = {{ .P2P.MaxNumOutboundPeers }}
  160. # Time to wait before flushing messages out on the connection
  161. flush_throttle_timeout = "{{ .P2P.FlushThrottleTimeout }}"
  162. # Maximum size of a message packet payload, in bytes
  163. max_packet_msg_payload_size = {{ .P2P.MaxPacketMsgPayloadSize }}
  164. # Rate at which packets can be sent, in bytes/second
  165. send_rate = {{ .P2P.SendRate }}
  166. # Rate at which packets can be received, in bytes/second
  167. recv_rate = {{ .P2P.RecvRate }}
  168. # Set true to enable the peer-exchange reactor
  169. pex = {{ .P2P.PexReactor }}
  170. # Seed mode, in which node constantly crawls the network and looks for
  171. # peers. If another node asks it for addresses, it responds and disconnects.
  172. #
  173. # Does not work if the peer-exchange reactor is disabled.
  174. seed_mode = {{ .P2P.SeedMode }}
  175. # Comma separated list of peer IDs to keep private (will not be gossiped to other peers)
  176. private_peer_ids = "{{ .P2P.PrivatePeerIDs }}"
  177. # Toggle to disable guard against peers connecting from the same ip.
  178. allow_duplicate_ip = {{ .P2P.AllowDuplicateIP }}
  179. # Peer connection configuration.
  180. handshake_timeout = "{{ .P2P.HandshakeTimeout }}"
  181. dial_timeout = "{{ .P2P.DialTimeout }}"
  182. ##### mempool configuration options #####
  183. [mempool]
  184. recheck = {{ .Mempool.Recheck }}
  185. broadcast = {{ .Mempool.Broadcast }}
  186. wal_dir = "{{ js .Mempool.WalPath }}"
  187. # Maximum number of transactions in the mempool
  188. size = {{ .Mempool.Size }}
  189. # Limit the total size of all txs in the mempool.
  190. # This only accounts for raw transactions (e.g. given 1MB transactions and
  191. # max_txs_bytes=5MB, mempool will only accept 5 transactions).
  192. max_txs_bytes = {{ .Mempool.MaxTxsBytes }}
  193. # Size of the cache (used to filter transactions we saw earlier) in transactions
  194. cache_size = {{ .Mempool.CacheSize }}
  195. ##### consensus configuration options #####
  196. [consensus]
  197. wal_file = "{{ js .Consensus.WalPath }}"
  198. timeout_propose = "{{ .Consensus.TimeoutPropose }}"
  199. timeout_propose_delta = "{{ .Consensus.TimeoutProposeDelta }}"
  200. timeout_prevote = "{{ .Consensus.TimeoutPrevote }}"
  201. timeout_prevote_delta = "{{ .Consensus.TimeoutPrevoteDelta }}"
  202. timeout_precommit = "{{ .Consensus.TimeoutPrecommit }}"
  203. timeout_precommit_delta = "{{ .Consensus.TimeoutPrecommitDelta }}"
  204. timeout_commit = "{{ .Consensus.TimeoutCommit }}"
  205. # Make progress as soon as we have all the precommits (as if TimeoutCommit = 0)
  206. skip_timeout_commit = {{ .Consensus.SkipTimeoutCommit }}
  207. # EmptyBlocks mode and possible interval between empty blocks
  208. create_empty_blocks = {{ .Consensus.CreateEmptyBlocks }}
  209. create_empty_blocks_interval = "{{ .Consensus.CreateEmptyBlocksInterval }}"
  210. # Reactor sleep duration parameters
  211. peer_gossip_sleep_duration = "{{ .Consensus.PeerGossipSleepDuration }}"
  212. peer_query_maj23_sleep_duration = "{{ .Consensus.PeerQueryMaj23SleepDuration }}"
  213. ##### transactions indexer configuration options #####
  214. [tx_index]
  215. # What indexer to use for transactions
  216. #
  217. # Options:
  218. # 1) "null"
  219. # 2) "kv" (default) - the simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend).
  220. indexer = "{{ .TxIndex.Indexer }}"
  221. # Comma-separated list of tags to index (by default the only tag is "tx.hash")
  222. #
  223. # You can also index transactions by height by adding "tx.height" tag here.
  224. #
  225. # It's recommended to index only a subset of tags due to possible memory
  226. # bloat. This is, of course, depends on the indexer's DB and the volume of
  227. # transactions.
  228. index_tags = "{{ .TxIndex.IndexTags }}"
  229. # When set to true, tells indexer to index all tags (predefined tags:
  230. # "tx.hash", "tx.height" and all tags from DeliverTx responses).
  231. #
  232. # Note this may be not desirable (see the comment above). IndexTags has a
  233. # precedence over IndexAllTags (i.e. when given both, IndexTags will be
  234. # indexed).
  235. index_all_tags = {{ .TxIndex.IndexAllTags }}
  236. ##### instrumentation configuration options #####
  237. [instrumentation]
  238. # When true, Prometheus metrics are served under /metrics on
  239. # PrometheusListenAddr.
  240. # Check out the documentation for the list of available metrics.
  241. prometheus = {{ .Instrumentation.Prometheus }}
  242. # Address to listen for Prometheus collector(s) connections
  243. prometheus_listen_addr = "{{ .Instrumentation.PrometheusListenAddr }}"
  244. # Maximum number of simultaneous connections.
  245. # If you want to accept a larger number than the default, make sure
  246. # you increase your OS limits.
  247. # 0 - unlimited.
  248. max_open_connections = {{ .Instrumentation.MaxOpenConnections }}
  249. # Instrumentation namespace
  250. namespace = "{{ .Instrumentation.Namespace }}"
  251. `
  252. /****** these are for test settings ***********/
  253. func ResetTestRoot(testName string) *Config {
  254. return ResetTestRootWithChainID(testName, "")
  255. }
  256. func ResetTestRootWithChainID(testName string, chainID string) *Config {
  257. // create a unique, concurrency-safe test directory under os.TempDir()
  258. rootDir, err := ioutil.TempDir("", fmt.Sprintf("%s-%s_", chainID, testName))
  259. if err != nil {
  260. panic(err)
  261. }
  262. // ensure config and data subdirs are created
  263. if err := cmn.EnsureDir(filepath.Join(rootDir, defaultConfigDir), DefaultDirPerm); err != nil {
  264. panic(err)
  265. }
  266. if err := cmn.EnsureDir(filepath.Join(rootDir, defaultDataDir), DefaultDirPerm); err != nil {
  267. panic(err)
  268. }
  269. baseConfig := DefaultBaseConfig()
  270. configFilePath := filepath.Join(rootDir, defaultConfigFilePath)
  271. genesisFilePath := filepath.Join(rootDir, baseConfig.Genesis)
  272. privKeyFilePath := filepath.Join(rootDir, baseConfig.PrivValidatorKey)
  273. privStateFilePath := filepath.Join(rootDir, baseConfig.PrivValidatorState)
  274. // Write default config file if missing.
  275. if !cmn.FileExists(configFilePath) {
  276. writeDefaultConfigFile(configFilePath)
  277. }
  278. if !cmn.FileExists(genesisFilePath) {
  279. if chainID == "" {
  280. chainID = "tendermint_test"
  281. }
  282. testGenesis := fmt.Sprintf(testGenesisFmt, chainID)
  283. cmn.MustWriteFile(genesisFilePath, []byte(testGenesis), 0644)
  284. }
  285. // we always overwrite the priv val
  286. cmn.MustWriteFile(privKeyFilePath, []byte(testPrivValidatorKey), 0644)
  287. cmn.MustWriteFile(privStateFilePath, []byte(testPrivValidatorState), 0644)
  288. config := TestConfig().SetRoot(rootDir)
  289. return config
  290. }
  291. var testGenesisFmt = `{
  292. "genesis_time": "2018-10-10T08:20:13.695936996Z",
  293. "chain_id": "%s",
  294. "validators": [
  295. {
  296. "pub_key": {
  297. "type": "tendermint/PubKeyEd25519",
  298. "value":"AT/+aaL1eB0477Mud9JMm8Sh8BIvOYlPGC9KkIUmFaE="
  299. },
  300. "power": "10",
  301. "name": ""
  302. }
  303. ],
  304. "app_hash": ""
  305. }`
  306. var testPrivValidatorKey = `{
  307. "address": "A3258DCBF45DCA0DF052981870F2D1441A36D145",
  308. "pub_key": {
  309. "type": "tendermint/PubKeyEd25519",
  310. "value": "AT/+aaL1eB0477Mud9JMm8Sh8BIvOYlPGC9KkIUmFaE="
  311. },
  312. "priv_key": {
  313. "type": "tendermint/PrivKeyEd25519",
  314. "value": "EVkqJO/jIXp3rkASXfh9YnyToYXRXhBr6g9cQVxPFnQBP/5povV4HTjvsy530kybxKHwEi85iU8YL0qQhSYVoQ=="
  315. }
  316. }`
  317. var testPrivValidatorState = `{
  318. "height": "0",
  319. "round": "0",
  320. "step": 0
  321. }`