Merge pull request #14 from Airstack-xyz/main

Main to Develop

Dockerfile

@@ -11,4 +11,6 @@ ENV TINI_VERSION v0.18.0
 ADD https://github.com/krallin/tini/releases/download/${TINI_VERSION}/tini /tini
 RUN chmod +x /tini
 
+EXPOSE 9000
+
 ENTRYPOINT ["/tini", "--", "python", "ethereumetl"]

blockchainetl/streaming/streamer.py

@@ -30,6 +30,7 @@
 from blockchainetl.file_utils import smart_open
 from ethereumetl.constants import constants
 
+from ethereumetl.metrics.prometheus import PrometheusConnector
 
 class Streamer:
     def __init__(
@@ -57,6 +58,10 @@ def __init__(
         self.mode = mode
         self.blocks_to_reprocess = blocks_to_reprocess
         self.last_synced_block = None
+
+        # init prometheus client
+        self.prometheus_client = PrometheusConnector()
+
         if self.mode == constants.RUN_MODE_NORMAL:
             if self.start_block is not None or not os.path.isfile(self.last_synced_block_file):
                 init_last_synced_block_file((self.start_block or 0) - 1, self.last_synced_block_file)
@@ -125,6 +130,11 @@ def _sync_cycle(self):
 
         logging.info('Current block {}, target block {}, last synced block {}, blocks to sync {}'.format(
             current_block, target_block, self.last_synced_block, blocks_to_sync))
+
+        self.prometheus_client.current_block.set(current_block)
+        self.prometheus_client.target_block.set(target_block)
+        self.prometheus_client.last_synced_block.set(self.last_synced_block)
+        self.prometheus_client.blocks_to_sync.set(blocks_to_sync)
 
         if blocks_to_sync != 0:
             self.blockchain_streamer_adapter.export_all(self.last_synced_block + 1, target_block)

ethereumetl/cli/stream.py

@@ -23,6 +23,7 @@
 import random
 import os
 import click
+from prometheus_client import start_http_server
 from blockchainetl.streaming.streaming_utils import configure_signals, configure_logging
 from ethereumetl.enumeration.entity_type import EntityType
 
@@ -72,7 +73,7 @@ def stream(last_synced_block_file, lag, output, start_block, end_block, entity_t
 
     if os.environ['KAFKA_BROKER_URI'] == None:
         raise ValueError('KAFKA_BROKER_URI env is missing')
-    
+
     if mode == constants.RUN_MODE_CORRECTION:
         blocks_to_reprocess = [int(block) for block in blocks_to_reprocess.split(',')]
         logging.info('blocks_to_reprocess: {} with length: {}'.format(blocks_to_reprocess, len(blocks_to_reprocess)))
@@ -98,6 +99,9 @@ def stream(last_synced_block_file, lag, output, start_block, end_block, entity_t
         mode=mode,
         blocks_to_reprocess=blocks_to_reprocess
     )
+
+    port = int(os.environ.get('METRICS_PORT', constants.METRICS_PORT))
+    start_http_server(port) # start prometheus server
     streamer.stream()
 
 

ethereumetl/constants/constants.py

@@ -40,4 +40,6 @@
 "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"]
 
 RUN_MODE_CORRECTION = 'correction'
-RUN_MODE_NORMAL= 'normal'
+RUN_MODE_NORMAL= 'normal'
+
+METRICS_PORT = '9000'

ethereumetl/metrics/prometheus.py

@@ -0,0 +1,8 @@
+from prometheus_client import Gauge
+
+class PrometheusConnector:
+    def __init__(self):
+        self.current_block = Gauge('current_block_height', 'Current block height')
+        self.target_block = Gauge('target_block_height', 'Target block height')
+        self.last_synced_block = Gauge('last_synced_block_height', 'Last synced block height')
+        self.blocks_to_sync = Gauge('blocks_to_sync', 'Number of Blocks To Sync')

setup.py

@@ -55,7 +55,8 @@ def read(fname):
             # that's why  we lock the version here
             'libcst==0.3.21',
             # Later versions break the build in Travis CI for Python 3.7.2
-            'grpcio==1.46.3'
+            'grpcio==1.46.3',
+            'prometheus-client==0.20.0'
         ],
         'streaming-kinesis': [
             'boto3==1.24.11',