Merge pull request #26 from p2p-org/parachains

Parachains

acala.env

@@ -0,0 +1,2 @@
+WS_ENDPOINT="wss://acala-rpc.dwellir.com"
+WS_ENDPOINTS="wss://acala-rpc.dwellir.com"

acala.yml

@@ -0,0 +1,17 @@
+version: '3.4'
+
+services:
+  acala_exporter:
+    build:
+      context: ./exporters/acala
+      args:
+        exporter: "acala_exporter"
+    environment:
+    - "LISTEN=0.0.0.0"
+    - "PORT=9150" 
+    - "CHAIN=acala"
+    env_file:
+    - ./acala.env
+    networks:
+    - exporters
+

exporters/acala/Dockerfile

@@ -1,10 +1,16 @@
-FROM --platform=linux/amd64 python:3.9.10-slim-buster
+FROM --platform=linux/amd64 python:3.11-slim-buster 
+
+ARG exporter
 
 WORKDIR /
 
+RUN apt-get update && apt-get install -y gcc g++
 COPY requirements.txt requirements.txt
-RUN pip3 install -r requirements.txt
+RUN pip3 install -r requirements.txt --no-cache-dir
+RUN groupadd -r exporter && useradd -r -g exporter exporter
 
 COPY exporter.py app.py
+COPY functions.py functions.py
 
+USER exporter
 CMD ["python3", "app.py"]

exporters/acala/config.yaml

@@ -1,5 +1,5 @@
 ---
-api_substrate: https://acala-api.polka.p2p.world/api
+ws_endpoint: ws://scaleway-acala-node1:9934
 chain: acala
 exporter:
   listen: 0.0.0.0

exporters/acala/exporter.py

@@ -6,15 +6,23 @@
 import time
 import json
 import logging
+import operator
+import traceback
+import os
 from collections import deque
 from flask import Flask, request, make_response
+from functions import SUBSTRATE_INTERFACE, get_era_points, get_chain_info 
+from _thread import interrupt_main
+from numpy import median, average, percentile
+from decimal import *
+
 
 logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO, datefmt='%Y-%m-%d %I:%M:%S')
 app = Flask(__name__)
 
 @app.route("/metrics")
 def metrics():
-    chain = get_config('chain')
+    chain = os.environ['CHAIN']
     metrics = q_metrics[0].copy()
 
     out = ""
@@ -51,64 +59,24 @@ def metrics():
 
     return response
 
-def api_request(endpoint = "api_substrate",method = None,args = None):
-    if endpoint == 'api_substrate':
-        url = get_config('api_substrate')
-
-        if isinstance(args, list):
-            for i in range(len(args)):
-                if isinstance(args[i], str):
-                    args[i] = '"' + args[i] + '"'
-        elif isinstance(args, str):
-            args = '"' + args + '"'
-        elif not args:
-            args = ""
-
-        data = {'method': method,
-                'args': args}
-
-    elif endpoint == 'api_registry':
-        url = get_config('api_registry')
-
-        data = {'method': method}
-
-    try:
-        r = requests.post(url, json=data)
-    except (ConnectionRefusedError,requests.exceptions.ConnectionError):
-        logging.critical('Coulnd not get data from ' + endpoint)
-
-        return None
-
-    if r.status_code == 200:
-        return r.json()['result']
-    else:
-        logging.critical('Request to ' + endpoint + ' finished with code ' + str(r.status_code))
-        return None
-
-def get_config(part):
-    with open('./config.yaml') as config_file:
-        data = yaml.load(config_file, Loader=yaml.FullLoader)
-
-    return data[part]
 
 def main():
     block = 0
     session = 0
 
     while True:
         try:
-            last_block = int(api_request(method = 'api.query.system.number'),16)
-
+            last_block = substrate_interface.request('System','Number').value
             if last_block != block:
-                validators = api_request(method = 'api.query.session.validators')
-                current_session = int(api_request(method = 'api.query.session.currentIndex'),16)
-                disabled_validators = api_request(method = 'api.query.session.disabledValidators')
+                validators = substrate_interface.request('Session','Validators').value
+                current_session = substrate_interface.request('Session','CurrentIndex').value
+                disabled_validators = substrate_interface.request('Session','DisabledValidators').value
                 result = {'validators':{},'common':{}}
                 result['common'] = {}
                 result['common']['active_validators_count'] = len(validators)
                 result['common']['current_session'] = current_session
                 for addr in validators: 
-                    points = int(api_request(method = 'api.query.collatorSelection.sessionPoints', args = addr),16)
+                    points =  substrate_interface.request('CollatorSelection','SessionPoints', [addr]).value
                     validator_points  = {k:points for k in validators if k == addr}
                     result['validators'].update(validator_points)
 
@@ -125,8 +93,14 @@ def main():
         time.sleep(3)
 
 if __name__ == '__main__':
-    endpoint_listen = get_config('exporter')['listen']
-    endpoint_port = get_config('exporter')['port']
+
+    endpoint_listen = os.environ['LISTEN']
+    endpoint_port = os.environ['PORT']
+    ws_endpoint = os.environ['WS_ENDPOINT']
+    chain = os.environ['CHAIN']
+
+
+    substrate_interface = SUBSTRATE_INTERFACE(ws_endpoint,chain)
 
     q_metrics = deque([])
 

exporters/acala/functions.py

@@ -0,0 +1,74 @@
+import yaml
+from substrateinterface import SubstrateInterface, Keypair
+from substrateinterface.exceptions import SubstrateRequestException
+from websocket._exceptions import WebSocketConnectionClosedException
+
+class SUBSTRATE_INTERFACE:
+    def __init__(self, ws_endpoint, chain):
+        self.substrate = SubstrateInterface(
+            url=ws_endpoint,
+            ss58_format=42,
+            type_registry_preset=chain)
+
+
+    def request(self, module: str, function: str, params: str = None):
+        try:
+            r = self.substrate.query(
+                module=module,
+                storage_function=function,
+                params=params)
+
+            return r
+        except (WebSocketConnectionClosedException,ConnectionRefusedError,SubstrateRequestException) as e:
+            self.substrate.connect_websocket()
+            logging.critical('The substrate api call failed with error ' + str(e))
+            r = None
+
+
+def get_era_points(data):
+    result = {}
+
+    for i in data['individual']:
+        result[i[0]] = i[1]
+
+    return {'result':result,'total':data['total']}
+
+def get_chain_info(chain,substrate_interface):
+    constants = {'polkadot':{'session_length':1200,'era_length':7200}}
+
+    session_length = constants[chain]['session_length']
+    era_length = constants[chain]['era_length']
+
+    current_era = substrate_interface.request('Staking','ActiveEra').value['index']
+    current_session = substrate_interface.request('Session','CurrentIndex').value
+
+    eras_start_session_index = substrate_interface.request('Staking','ErasStartSessionIndex',[current_era]).value
+
+    genesis_slot = substrate_interface.request('Babe','GenesisSlot').value
+    current_slot = substrate_interface.request('Babe','CurrentSlot').value
+
+    session_start_slot = int(current_session) * int(session_length) + int(genesis_slot)
+    session_progress = int(current_slot) - int(session_start_slot)
+
+    era_session_index = int(current_session) - int(eras_start_session_index)
+    era_progress = int(era_session_index) * int(session_length) + int(session_progress)
+
+    return {'current_era': current_era, 'eras_start_session_index': eras_start_session_index, 'current_session': current_session, 'era_progress': era_progress / era_length * 100, 'session_progress': session_progress / session_length * 100}
+
+def ss58_convert(data):
+    r = []
+
+    for key in data:
+        pubkey = Keypair(ss58_address=key).public_key.hex()
+        r.append('0x' + pubkey)
+
+    return r
+
+def get_keys(validators,keys):
+    result = {i:None for i in validators}
+
+    for i in keys:
+        if str(i[0]) in result.keys():
+            result[str(i[0])] = str(i[1]['grandpa'])
+
+    return result

exporters/acala/requirements.txt

@@ -1,3 +1,4 @@
-flask
-requests
-pyyaml
+pyaml==23.5.9
+substrate-interface==1.7.1
+flask==2.3.2
+numpy==1.24.3