kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'philip/improving-import-scripts-logging' into 'master'

Browse Source 
Improves import scripts logging

Closes #74

See merge request grassrootseconomics/cic-internal-integration!207
This commit is contained in:
Blair Vanderlugt 2021-07-03 16:55:51 +00:00
commit a942c785f6
7 changed files with 202 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
apps/cic-eth/test_requirements.txt
View File

@ -4,3 +4,4 @@ pytest-mock==3.3.1
pytest-cov==2.10.1
eth-tester==0.5.0b3
py-evm==0.3.0a20
eth-erc20==0.0.10a2

2
apps/cic-ussd/cic_ussd/runnable/daemons/cic_user_tasker.py
View File

@ -127,6 +127,8 @@ def main():
argv.append('--loglevel=INFO')
argv.append('-Q')
argv.append(args.q)
argv.append('-n')
argv.append(args.q)
current_app.worker_main(argv)

61
apps/data-seeding/cic_ussd/import_balance.py
View File

@ -1,28 +1,22 @@
# standard imports
import os
import sys
import logging
import argparse
import hashlib
import redis
import celery
import logging
import sys
import os
# external imports
import celery
import confini
from chainlib.eth.connection import EthHTTPConnection
import redis
from chainlib.chain import ChainSpec
from hexathon import (
strip_0x,
add_0x,
)
from chainlib.eth.address import to_checksum_address
from chainlib.eth.connection import EthHTTPConnection
from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
from crypto_dev_signer.keystore.dict import DictKeystore
from cic_types.models.person import Person
# local imports
from import_util import BalanceProcessor
from import_task import *
from import_task import ImportTask, MetadataTask
from import_util import BalanceProcessor, get_celery_worker_status
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
@ -41,25 +35,29 @@ argparser.add_argument('--meta-port', dest='meta_port', type=int, help='metadata
argparser.add_argument('--redis-host', dest='redis_host', type=str, help='redis host to use for task submission')
argparser.add_argument('--redis-port', dest='redis_port', type=int, help='redis host to use for task submission')
argparser.add_argument('--redis-db', dest='redis_db', type=int, help='redis db to use for task submission and callback')
argparser.add_argument('--token-symbol', default='SRF', type=str, dest='token_symbol', help='Token symbol to use for trnsactions')
argparser.add_argument('--token-symbol', default='GFT', type=str, dest='token_symbol',
help='Token symbol to use for transactions')
argparser.add_argument('--head', action='store_true', help='start at current block height (overrides --offset)')
argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
argparser.add_argument('-q', type=str, default='cic-eth', help='celery queue to submit transaction tasks to')
argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str,
help='environment prefix for variables to overwrite configuration')
argparser.add_argument('-q', type=str, default='cic-import-ussd', help='celery queue to submit transaction tasks to')
argparser.add_argument('--offset', type=int, default=0, help='block offset to start syncer from')
argparser.add_argument('-v', help='be verbose', action='store_true')
argparser.add_argument('-vv', help='be more verbose', action='store_true')
argparser.add_argument('user_dir', default='out', type=str, help='user export directory')
args = argparser.parse_args(sys.argv[1:])
if args.v == True:
if args.v:
logging.getLogger().setLevel(logging.INFO)
elif args.vv == True:
elif args.vv:
logging.getLogger().setLevel(logging.DEBUG)
config_dir = os.path.join(args.c)
os.makedirs(config_dir, 0o777, True)
config = confini.Config(config_dir, args.env_prefix)
config.process()
# override args
args_override = {
'CIC_CHAIN_SPEC': getattr(args, 'i'),
@ -81,14 +79,19 @@ redis_host = config.get('REDIS_HOST')
redis_port = config.get('REDIS_PORT')
redis_db = config.get('REDIS_DB')
r = redis.Redis(redis_host, redis_port, redis_db)
# create celery apps
celery_app = celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY_BROKER_URL'))
status = get_celery_worker_status(celery_app=celery_app)
signer_address = None
keystore = DictKeystore()
if args.y != None:
if args.y is not None:
logg.debug('loading keystore file {}'.format(args.y))
signer_address = keystore.import_keystore_file(args.y)
logg.debug('now have key for signer address {}'.format(signer_address))
# define signer
signer = EIP155Signer(keystore)
queue = args.q
@ -111,10 +114,12 @@ MetadataTask.meta_host = config.get('META_HOST')
MetadataTask.meta_port = config.get('META_PORT')
ImportTask.chain_spec = chain_spec
def main():
conn = EthHTTPConnection(config.get('ETH_PROVIDER'))
ImportTask.balance_processor = BalanceProcessor(conn, chain_spec, config.get('CIC_REGISTRY_ADDRESS'), signer_address, signer)
ImportTask.balance_processor = BalanceProcessor(conn, chain_spec, config.get('CIC_REGISTRY_ADDRESS'),
signer_address, signer)
ImportTask.balance_processor.init(token_symbol)
# TODO get decimals from token
@ -124,7 +129,7 @@ def main():
i = 0
while True:
l = f.readline()
if l == None:
if l is None:
break
r = l.split(',')
try:
@ -147,11 +152,19 @@ def main():
[
MetadataTask.balance_processor.nonce_offset,
],
queue='cic-import-ussd',
queue=queue,
)
s.apply_async()
argv = ['worker', '-Q', 'cic-import-ussd', '--loglevel=DEBUG']
argv = ['worker']
if args.vv:
argv.append('--loglevel=DEBUG')
elif args.v:
argv.append('--loglevel=INFO')
argv.append('-Q')
argv.append(args.q)
argv.append('-n')
argv.append(args.q)
celery_app.worker_main(argv)

8
apps/data-seeding/cic_ussd/import_pins.py
View File

@ -9,7 +9,7 @@ import celery
import confini
# local imports
from import_task import *
from import_util import get_celery_worker_status
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
@ -39,9 +39,12 @@ elif args.vv:
config_dir = args.c
config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
config.process()
config.censor('PASSWORD', 'DATABASE')
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
status = get_celery_worker_status(celery_app=celery_app)
db_configs = {
'database': config.get('DATABASE_NAME'),
@ -61,7 +64,8 @@ def main():
(db_configs, phone_to_pins),
queue=args.q
)
s_import_pins.apply_async()
result = s_import_pins.apply_async()
logg.debug(f'TASK: {result.id}, STATUS: {result.status}')
if __name__ == '__main__':

39
apps/data-seeding/cic_ussd/import_task.py
View File

@ -1,36 +1,33 @@
# standard imports
import os
import logging
import random
import urllib.parse
import urllib.error
import urllib.request
import json
import logging
import os
import random
import urllib.error
import urllib.parse
import urllib.request
# external imports
import celery
import psycopg2
from psycopg2 import extras
from hexathon import (
strip_0x,
add_0x,
)
from chainlib.eth.address import to_checksum_address
from chainlib.eth.tx import (
unpack,
raw,
)
from cic_types.processor import generate_metadata_pointer
from cic_types.models.person import Person
from cic_types.processor import generate_metadata_pointer
from hexathon import (
strip_0x,
add_0x,
)
#logg = logging.getLogger().getChild(__name__)
logg = logging.getLogger()
celery_app = celery.current_app
class ImportTask(celery.Task):
balances = None
import_dir = 'out'
count = 0
@ -38,8 +35,8 @@ class ImportTask(celery.Task):
balance_processor = None
max_retries = None
class MetadataTask(ImportTask):
class MetadataTask(ImportTask):
meta_host = None
meta_port = None
meta_path = ''
@ -180,8 +177,6 @@ def generate_metadata(self, address, phone):
@celery_app.task(bind=True, base=MetadataTask)
def opening_balance_tx(self, address, phone, serial):
old_address = old_address_from_phone(self.import_dir, phone)
k = to_checksum_address(strip_0x(old_address))
@ -215,14 +210,14 @@ def opening_balance_tx(self, address, phone, serial):
return tx['hash']
@celery_app.task(bind=True, base=ImportTask, autoretry_for=(FileNotFoundError,), max_retries=None, default_retry_delay=0.1)
@celery_app.task(bind=True, base=ImportTask, autoretry_for=(FileNotFoundError,), max_retries=None,
default_retry_delay=0.1)
def send_txs(self, nonce):
if nonce == self.count + self.balance_processor.nonce_offset:
logg.info('reached nonce {} (offset {} + count {}) exiting'.format(nonce, self.balance_processor.nonce_offset, self.count))
logg.info('reached nonce {} (offset {} + count {}) exiting'.format(nonce, self.balance_processor.nonce_offset,
self.count))
return
logg.debug('attempt to open symlink for nonce {}'.format(nonce))
tx_nonce_path = os.path.join(
self.import_dir,
@ -252,7 +247,6 @@ def send_txs(self, nonce):
)
s.apply_async()
return nonce
@ -310,4 +304,3 @@ def set_ussd_data(config: dict, ussd_data: dict):
# close connections
db_cursor.close()
db_conn.close()

31
apps/data-seeding/cic_ussd/import_users.py
View File

@ -17,6 +17,9 @@ import redis
from chainlib.chain import ChainSpec
from cic_types.models.person import Person
# local imports
from import_util import get_celery_worker_status
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
@ -28,11 +31,14 @@ argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='Chain spe
argparser.add_argument('--redis-host', dest='redis_host', type=str, help='redis host to use for task submission')
argparser.add_argument('--redis-port', dest='redis_port', type=int, help='redis host to use for task submission')
argparser.add_argument('--redis-db', dest='redis_db', type=int, help='redis db to use for task submission and callback')
argparser.add_argument('--batch-size', dest='batch_size', default=100, type=int, help='burst size of sending transactions to node') # batch size should be slightly below cumulative gas limit worth, eg 80000 gas txs with 8000000 limit is a bit less than 100 batch size
argparser.add_argument('--batch-size', dest='batch_size', default=100, type=int,
help='burst size of sending transactions to node') # batch size should be slightly below cumulative gas limit worth, eg 80000 gas txs with 8000000 limit is a bit less than 100 batch size
argparser.add_argument('--batch-delay', dest='batch_delay', default=3, type=int, help='seconds delay between batches')
argparser.add_argument('--timeout', default=60.0, type=float, help='Callback timeout')
argparser.add_argument('--ussd-host', dest='ussd_host', type=str, help="host to ussd app responsible for processing ussd requests.")
argparser.add_argument('--ussd-port', dest='ussd_port', type=str, help="port to ussd app responsible for processing ussd requests.")
argparser.add_argument('--ussd-host', dest='ussd_host', type=str,
help="host to ussd app responsible for processing ussd requests.")
argparser.add_argument('--ussd-port', dest='ussd_port', type=str,
help="port to ussd app responsible for processing ussd requests.")
argparser.add_argument('--ussd-no-ssl', dest='ussd_no_ssl', help='do not use ssl (careful)', action='store_true')
argparser.add_argument('-q', type=str, default='cic-eth', help='Task queue')
argparser.add_argument('-v', action='store_true', help='Be verbose')
@ -55,7 +61,10 @@ args_override = {
'REDIS_DB': getattr(args, 'redis_db'),
}
config.dict_override(args_override, 'cli')
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
get_celery_worker_status(celery_app=celery_app)
redis_host = config.get('REDIS_HOST')
redis_port = config.get('REDIS_PORT')
@ -65,22 +74,22 @@ r = redis.Redis(redis_host, redis_port, redis_db)
ps = r.pubsub()
user_new_dir = os.path.join(args.user_dir, 'new')
os.makedirs(user_new_dir)
os.makedirs(user_new_dir, exist_ok=True)
ussd_data_dir = os.path.join(args.user_dir, 'ussd')
os.makedirs(ussd_data_dir)
os.makedirs(ussd_data_dir, exist_ok=True)
preferences_dir = os.path.join(args.user_dir, 'preferences')
os.makedirs(os.path.join(preferences_dir, 'meta'))
os.makedirs(os.path.join(preferences_dir, 'meta'), exist_ok=True)
meta_dir = os.path.join(args.user_dir, 'meta')
os.makedirs(meta_dir)
os.makedirs(meta_dir, exist_ok=True)
user_old_dir = os.path.join(args.user_dir, 'old')
os.stat(user_old_dir)
txs_dir = os.path.join(args.user_dir, 'txs')
os.makedirs(txs_dir)
os.makedirs(txs_dir, exist_ok=True)
chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
chain_str = str(chain_spec)
@ -95,6 +104,7 @@ if ussd_no_ssl is True:
else:
ussd_ssl = True
def build_ussd_request(phone, host, port, service_code, username, password, ssl=False):
url = 'http'
if ssl:
@ -115,7 +125,7 @@ def build_ussd_request(phone, host, port, service_code, username, password, ssl=
'text': service_code,
}
req = urllib.request.Request(url)
req.method=('POST')
req.method = 'POST'
data_str = urlencode(data)
data_bytes = data_str.encode('utf-8')
req.add_header('Content-Type', 'application/x-www-form-urlencoded')
@ -164,7 +174,7 @@ if __name__ == '__main__':
f.close()
u = Person.deserialize(o)
new_address = register_ussd(i, u)
register_ussd(i, u)
phone_object = phonenumbers.parse(u.tel)
phone = phonenumbers.format_number(phone_object, phonenumbers.PhoneNumberFormat.E164)
@ -206,4 +216,3 @@ if __name__ == '__main__':
if j == batch_size:
time.sleep(batch_delay)
j = 0

24
apps/data-seeding/cic_ussd/import_util.py
View File

@ -2,15 +2,16 @@
import logging
# external imports
from eth_contract_registry import Registry
from eth_token_index import TokenUniqueSymbolIndex
from chainlib.eth.gas import OverrideGasOracle
from chainlib.eth.nonce import OverrideNonceOracle
from eth_erc20 import ERC20
from chainlib.eth.tx import (
count,
TxFormat,
)
from celery import Celery
from eth_contract_registry import Registry
from eth_erc20 import ERC20
from eth_token_index import TokenUniqueSymbolIndex
logg = logging.getLogger().getChild(__name__)
@ -18,7 +19,6 @@ logg = logging.getLogger().getChild(__name__)
class BalanceProcessor:
def __init__(self, conn, chain_spec, registry_address, signer_address, signer):
self.chain_spec = chain_spec
self.conn = conn
# self.signer_address = signer_address
@ -36,7 +36,6 @@ class BalanceProcessor:
self.value_multiplier = 1
def init(self, token_symbol):
# Get Token registry address
registry = Registry(self.chain_spec)
@ -57,16 +56,25 @@ class BalanceProcessor:
n = tx_factory.parse_decimals(r)
self.value_multiplier = 10 ** n
def get_rpc_tx(self, recipient, value, i):
logg.debug('initiating nonce offset {} for recipient {}'.format(self.nonce_offset + i, recipient))
nonce_oracle = OverrideNonceOracle(self.signer_address, self.nonce_offset + i)
tx_factory = ERC20(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=self.gas_oracle)
return tx_factory.transfer(self.token_address, self.signer_address, recipient, value, tx_format=TxFormat.RLP_SIGNED)
return tx_factory.transfer(self.token_address, self.signer_address, recipient, value,
tx_format=TxFormat.RLP_SIGNED)
# (tx_hash_hex, o) = tx_factory.transfer(self.token_address, self.signer_address, recipient, value)
# self.conn.do(o)
# return tx_hash_hex
def get_decimal_amount(self, value):
return value * self.value_multiplier
def get_celery_worker_status(celery_app: Celery):
inspector = celery_app.control.inspect()
availability = inspector.ping()
status = {
'availability': availability,
}
logg.debug(f'RUNNING WITH STATUS: {status}')
return status