[WIP] feat: separate performance collection and distribution

pyproject.toml

@@ -30,6 +30,11 @@ oz-merkle-tree = { git = "https://github.com/lidofinance/oz-merkle-tree", rev =
 multiformats = "^0.3.1"
 protobuf="^6.31.1"
 dag-cbor="^0.3.3"
+fastapi = "^0.121.3"
+uvicorn = "^0.38.0"
+sqlmodel = "^0.0.27"
+psycopg2-binary = "^2.9.11"
+prometheus-fastapi-instrumentator = "^7.1.0"
 
 [tool.poetry.group.dev.dependencies]
 base58 = "^2.1.1"

src/constants.py

@@ -33,6 +33,7 @@
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/beacon-chain.md#time-parameters
 SLOTS_PER_HISTORICAL_ROOT = 2**13  # 8192
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/altair/beacon-chain.md#sync-committee
+SYNC_COMMITTEE_SIZE = 512
 EPOCHS_PER_SYNC_COMMITTEE_PERIOD = 256
 # https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/beacon-chain.md#domain-types
 DOMAIN_DEPOSIT_TYPE = bytes.fromhex("03000000")  # 0x03000000

src/main.py

@@ -15,6 +15,7 @@
 from src.modules.csm.csm import CSOracle
 from src.modules.ejector.ejector import Ejector
 from src.providers.ipfs import IPFSProvider, Kubo, LidoIPFS, Pinata, Storacha
+from src.modules.performance.collector.collector import PerformanceCollector
 from src.types import OracleModule
 from src.utils.build import get_build_info
 from src.utils.exception import IncompatibleException
@@ -29,6 +30,7 @@
     LidoValidatorsProvider,
     TransactionUtils,
 )
+from src.web3py.extensions.performance import PerformanceClientModule
 from src.web3py.types import Web3
 from decimal import getcontext
 
@@ -75,6 +77,9 @@ def main(module_name: OracleModule):
     logger.info({'msg': 'Initialize IPFS providers.'})
     ipfs = IPFS(web3, ipfs_providers(), retries=variables.HTTP_REQUEST_RETRY_COUNT_IPFS)
 
+    logger.info({'msg': 'Initialize Performance Collector client.'})
+    performance = PerformanceClientModule(variables.PERFORMANCE_COLLECTOR_URI)
+
     logger.info({'msg': 'Check configured providers.'})
     if Version(kac.get_status().appVersion) < constants.ALLOWED_KAPI_VERSION:
         raise IncompatibleException(f'Incompatible KAPI version. Required >= {constants.ALLOWED_KAPI_VERSION}.')
@@ -89,12 +94,13 @@ def main(module_name: OracleModule):
         'cc': lambda: cc,  # type: ignore[dict-item]
         'kac': lambda: kac,  # type: ignore[dict-item]
         'ipfs': lambda: ipfs,  # type: ignore[dict-item]
+        'performance': lambda: performance,  # type: ignore[dict-item]
     })
 
     logger.info({'msg': 'Initialize prometheus metrics.'})
     init_metrics()
 
-    instance: Accounting | Ejector | CSOracle
+    instance: Accounting | Ejector | CSOracle | PerformanceCollector
     if module_name == OracleModule.ACCOUNTING:
         logger.info({'msg': 'Initialize Accounting module.'})
         instance = Accounting(web3)
@@ -104,10 +110,15 @@ def main(module_name: OracleModule):
     elif module_name == OracleModule.CSM:
         logger.info({'msg': 'Initialize CSM performance oracle module.'})
         instance = CSOracle(web3)
+    elif module_name == OracleModule.PERFORMANCE_COLLECTOR:
+        logger.info({'msg': 'Initialize Performance Collector module.'})
+        # FIXME: web3 object is overkill. only CONSENSUS_CLIENT_URI needed here.
+        instance = PerformanceCollector(web3)
     else:
         raise ValueError(f'Unexpected arg: {module_name=}.')
 
-    instance.check_contract_configs()
+    if module_name != OracleModule.PERFORMANCE_COLLECTOR:
+        instance.check_contract_configs()
 
     if variables.DAEMON:
         instance.run_as_daemon()
@@ -174,7 +185,6 @@ def ipfs_providers() -> Iterator[IPFSProvider]:
         )
 
 
-
 if __name__ == '__main__':
     module_name_arg = sys.argv[-1]
     if module_name_arg not in OracleModule:
@@ -183,12 +193,22 @@ def ipfs_providers() -> Iterator[IPFSProvider]:
         raise ValueError(msg)
 
     module = OracleModule(module_name_arg)
+
     if module is OracleModule.CHECK:
         errors = variables.check_uri_required_variables()
         variables.raise_from_errors(errors)
-
         sys.exit(check())
 
-    errors = variables.check_all_required_variables(module)
+    if module is OracleModule.PERFORMANCE_WEB_SERVER:
+        from src.modules.performance.web.server import serve
+        errors = variables.check_perf_web_server_required_variables()
+        variables.raise_from_errors(errors)
+        logger.info({'msg': f'Starting Performance Web Server on port {variables.PERFORMANCE_WEB_SERVER_API_PORT}'})
+        sys.exit(serve())
+
+    if module is OracleModule.PERFORMANCE_COLLECTOR:
+        errors = variables.check_perf_collector_required_variables()
+    else:
+        errors = variables.check_all_required_variables(module)
     variables.raise_from_errors(errors)
     main(module)

src/metrics/prometheus/basic.py

@@ -68,6 +68,14 @@ class Status(Enum):
     buckets=requests_buckets,
 )
 
+PERFORMANCE_REQUESTS_DURATION = Histogram(
+    'performance_requests_duration',
+    'Duration of requests to Performance Collector API',
+    ['endpoint', 'code', 'domain'],
+    namespace=PROMETHEUS_PREFIX,
+    buckets=requests_buckets,
+)
+
 KEYS_API_REQUESTS_DURATION = Histogram(
     'keys_api_requests_duration',
     'Duration of requests to Keys API',