Merge pull request #340 from skalenetwork/statsd

Introduce statsd

requirements.txt

@@ -1 +1,2 @@
-skale.py==6.2b0
+skale.py==6.2b0
+statsd==4.0.1

transaction_manager/attempt_manager/v2.py

@@ -35,6 +35,7 @@
     MIN_PRIORITY_FEE
 )
 from ..eth import Eth
+from ..resources import stdc
 from ..structures import Attempt, Fee, Tx
 
 logger = logging.getLogger(__name__)
@@ -165,8 +166,11 @@ def make(self, tx: Tx) -> None:
                 last.fee.max_fee_per_gas,  # type: ignore
                 min_fee=estimated_base_fee
             )
+            stdc.gauge('tm.max_priority_fee', tip)
+            stdc.gauge('tm.max_fee_per_gas', gap)
             next_fee = Fee(max_priority_fee_per_gas=tip, max_fee_per_gas=gap)
             next_wait_time = self.next_waiting_time(next_index)
+            stdc.gauge('tm.next_wating_time', gap)
 
         logger.info('Next fee %s', next_fee)
         tx.fee, tx.nonce = next_fee, nonce

transaction_manager/config.py

@@ -46,6 +46,8 @@
 DEFAULT_ID_LEN = 19
 DEFAULT_GAS_LIMIT: int = 1000000
 IMA_ID_SUFFIX = 'js'
+STATSD_HOST: str = '127.0.0.1'
+STATSD_PORT: int = 8125
 
 # V1
 AVG_GAS_PRICE_INC_PERCENT = 50

transaction_manager/eth.py

@@ -36,7 +36,7 @@
     MAX_WAITING_TIME,
     TARGET_REWARD_PERCENTILE
 )
-from .resources import w3 as gw3
+from .resources import stdc, w3 as gw3
 from .structures import Tx
 
 logger = logging.getLogger(__name__)
@@ -83,7 +83,8 @@ def block_gas_limit(self) -> int:
 
     @cached_property
     def chain_id(self) -> int:
-        return self.w3.eth.chain_id
+        with stdc.timer('tm.chain_id_request'):
+            return self.w3.eth.chain_id
 
     def get_balance(self, address: str) -> int:
         checksum_addres = self.w3.to_checksum_address(address)

transaction_manager/main.py

@@ -41,6 +41,7 @@ def run_proc():
         RedisAttemptStorage(),
         wallet.address
     )
+
     proc = Processor(eth, pool, attempt_manager, wallet)
     logger.info('Starting transaction processor')
     proc.run()

transaction_manager/processor.py

@@ -32,6 +32,7 @@
     is_replacement_underpriced,
     ReceiptTimeoutError
 )
+from .resources import stdc
 from .structures import Tx, TxStatus
 from .txpool import TxPool
 
@@ -191,6 +192,8 @@ def acquire_tx(self, tx: Tx) -> Generator[Tx, None, None]:
         try:
             yield tx
         finally:
+            stdc.gauge('tm.attempt', tx.attempts)
+            stdc.incr(f'tm.transaction.{tx.status.name}')
             if tx.is_sent():
                 self.attempt_manager.save()
             if not tx.is_completed() and tx.is_last_attempt():
@@ -210,11 +213,13 @@ def process_next(self) -> None:
             with self.acquire_tx(tx) as tx:
                 logger.info(
                     'Previous attempt %s', self.attempt_manager.current)
-                self.process(tx)
+                with stdc.timer('tm.transaction.time'):
+                    self.process(tx)
 
     def run(self) -> None:
         while True:
             try:
+                stdc.gauge('tm.pool.size', self.pool.size)
                 self.process_next()
             except Exception:
                 logger.exception('Failed to process tx')

transaction_manager/resources.py

@@ -18,12 +18,14 @@
 #   along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import redis
+import statsd  # type: ignore
 
 from skale.utils.web3_utils import init_web3  # type: ignore
 from web3 import Web3
 
-from .config import ALLOWED_TS_DIFF, ENDPOINT, REDIS_URI
+from .config import ALLOWED_TS_DIFF, ENDPOINT, REDIS_URI, STATSD_HOST, STATSD_PORT
 
 cpool: redis.ConnectionPool = redis.ConnectionPool.from_url(REDIS_URI)
 rs: redis.Redis = redis.Redis(connection_pool=cpool)
 w3: Web3 = init_web3(ENDPOINT, ts_diff=ALLOWED_TS_DIFF)
+stdc: statsd.StatsClient = statsd.StatsClient(STATSD_HOST, STATSD_PORT)