Merge remote-tracking branch 'origin/master' into lash/import-ussd

apps/cic-cache/requirements.txt

@@ -1,13 +1,13 @@
-cic-base~=0.1.2a58
+cic-base~=0.1.2a62
 alembic==1.4.2
 confini~=0.3.6rc3
 uwsgi==2.0.19.1
 moolb~=0.1.0
-cic-eth-registry~=0.5.4a10
+cic-eth-registry~=0.5.4a12
 SQLAlchemy==1.3.20
 semver==2.13.0
 psycopg2==2.8.6
 celery==4.4.7
 redis==3.5.3
-chainlib~=0.0.2a2
+chainlib~=0.0.2a5
 chainsyncer~=0.0.1a21

apps/cic-eth/cic_eth/runnable/daemons/retry.py

@@ -20,6 +20,7 @@ from cic_eth.admin.ctrl import lock_send
 from cic_eth.db.enum import LockEnum
 from cic_eth.runnable.daemons.filters.straggler import StragglerFilter
 from cic_eth.sync.retry import RetrySyncer
+from cic_eth.stat import init_chain_stat
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -71,57 +72,21 @@ RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, tag='def
 dsn = dsn_from_config(config)
 SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
 
-straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
-
-## TODO: we already have the signed raw tx in get, so its a waste of cycles to get_tx here
-#def sendfail_filter(w3, tx_hash, rcpt, chain_spec):
-#    tx_dict = get_tx(tx_hash)
-#    tx = unpack(tx_dict['signed_tx'], chain_spec)
-#    logg.debug('submitting tx {} for retry'.format(tx_hash))
-#    s_check = celery.signature(
-#            'cic_eth.admin.ctrl.check_lock',
-#            [
-#                tx_hash,
-#                chain_str,
-#                LockEnum.QUEUE,
-#                tx['from'],
-#                ],
-#            queue=queue,
-#            )
-##    s_resume = celery.signature(
-##            'cic_eth.eth.tx.resume_tx',
-##            [
-##                chain_str,
-##                ],
-##            queue=queue,
-##            )
-#    
-##    s_retry_status = celery.signature(
-##            'cic_eth.queue.state.set_ready',
-##            [],
-##            queue=queue,
-##    )
-#    s_resend = celery.signature(
-#            'cic_eth.eth.gas.resend_with_higher_gas',
-#            [
-#                chain_str,
-#                ],
-#            queue=queue,
-#            )
-#
-#    #s_resume.link(s_retry_status)
-#    #s_check.link(s_resume)
-#    s_check.link(s_resend)
-#    s_check.apply_async()
-
 
 def main(): 
     conn = RPCConnection.connect(chain_spec, 'default')
+
+    straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
+    loop_interval = config.get('SYNCER_LOOP_INTERVAL')
+    if loop_interval == None:
+        stat = init_chain_stat(conn)
+        loop_interval = stat.block_average()
+
     syncer = RetrySyncer(conn, chain_spec, straggler_delay, batch_size=config.get('_BATCH_SIZE'))
     syncer.backend.set(0, 0)
     fltr = StragglerFilter(chain_spec, queue=queue)
     syncer.add_filter(fltr)
-    syncer.loop(float(straggler_delay), conn)
+    syncer.loop(int(loop_interval), conn)
 
 
 if __name__ == '__main__':

apps/cic-eth/cic_eth/runnable/daemons/tracker.py

@@ -7,7 +7,7 @@ import argparse
 import sys
 import re
 
-# third-party imports
+# external imports
 import confini
 import celery
 import rlp
@@ -42,6 +42,7 @@ from cic_eth.runnable.daemons.filters import (
         RegistrationFilter,
         TransferAuthFilter,
         )
+from cic_eth.stat import init_chain_stat
 
 script_dir = os.path.realpath(os.path.dirname(__file__))
 
@@ -78,6 +79,11 @@ def main():
     block_current = int(r, 16)
     block_offset = block_current + 1
 
+    loop_interval = config.get('SYNCER_LOOP_INTERVAL')
+    if loop_interval == None:
+        stat = init_chain_stat(rpc, block_start=block_current)
+        loop_interval = stat.block_average()
+
     logg.debug('starting at block {}'.format(block_offset))
 
     syncers = []
@@ -140,7 +146,8 @@ def main():
         for cf in callback_filters:
             syncer.add_filter(cf)
 
-        r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        #r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        r = syncer.loop(int(loop_interval), rpc)
         sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
 
         i += 1

apps/cic-eth/cic_eth/stat.py

@@ -0,0 +1,33 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.stat import ChainStat
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+
+logg = logging.getLogger().getChild(__name__)
+
+BLOCK_SAMPLES = 10
+
+
+def init_chain_stat(rpc, block_start=0):
+    stat = ChainStat()
+
+    if block_start == 0:
+        o = block_latest()
+        r = rpc.do(o)
+        block_start = int(r, 16)
+
+    for i in range(BLOCK_SAMPLES):
+        o = block_by_number(block_start-10+i)
+        block_src = rpc.do(o)
+        logg.debug('block {}'.format(block_src))
+        block = Block(block_src)
+        stat.block_apply(block)
+
+    logg.debug('calculated block time {} from {} block samples'.format(stat.block_average(), BLOCK_SAMPLES))
+    return stat

apps/cic-eth/config/docker/cic.ini

@@ -2,3 +2,4 @@
 registry_address = 
 chain_spec = evm:bloxberg:8996
 trust_address = 0xEb3907eCad74a0013c259D5874AE7f22DcBcC95C
+tx_retry_delay = 20

apps/cic-eth/config/docker/syncer.ini

@@ -1,2 +1,2 @@
 [SYNCER]
-loop_interval = 1
+loop_interval = 

apps/cic-eth/config/syncer.ini

@@ -1,2 +1,2 @@
 [SYNCER]
-loop_interval = 1
+loop_interval = 

apps/cic-eth/docker/Dockerfile

@@ -29,7 +29,7 @@ RUN /usr/local/bin/python -m pip install --upgrade pip
 #	python merge_requirements.py | tee merged_requirements.txt 
 #RUN cd cic-base && \
 #	pip install $pip_extra_index_url_flag -r ./merged_requirements.txt 
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a61
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a62
 
 COPY cic-eth/scripts/ scripts/
 COPY cic-eth/setup.cfg cic-eth/setup.py ./

apps/cic-eth/requirements.txt

@@ -1,4 +1,4 @@
-cic-base~=0.1.2a60
+cic-base~=0.1.2a62
 celery==4.4.7
 crypto-dev-signer~=0.4.14a17
 confini~=0.3.6rc3
@@ -16,7 +16,7 @@ semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.1a7
-chainlib~=0.0.2a4
+chainlib~=0.0.2a5
 hexathon~=0.0.1a7
 chainsyncer~=0.0.1a21
 chainqueue~=0.0.1a7

apps/cic-notify/cic_notify/api.py

@@ -3,6 +3,7 @@ import logging
 import re
 
 # third-party imports
+from celery.app.control import Inspect
 import celery
 
 # local imports
@@ -15,6 +16,29 @@ logg = logging.getLogger()
 sms_tasks_matcher = r"^(cic_notify.tasks.sms)(\.\w+)?"
 
 
+re_q = r'^cic-notify'
+def get_sms_queue_tasks(app, task_prefix='cic_notify.tasks.sms.'):
+    host_queues = []
+
+    i = Inspect(app=app)
+    qs = i.active_queues()
+    for host in qs.keys():
+        for q in qs[host]:
+            if re.match(re_q, q['name']):
+                host_queues.append((host, q['name'],))
+   
+    task_prefix_len = len(task_prefix)
+    queue_tasks = []
+    for (host, queue) in host_queues:
+        i = Inspect(app=app, destination=[host])
+        for tasks in i.registered_tasks().values():
+            for task in tasks:
+                if len(task) >= task_prefix_len and task[:task_prefix_len] == task_prefix:
+                    queue_tasks.append((queue, task,))
+    
+    return queue_tasks
+
+
 class Api:
     # TODO: Implement callback strategy
     def __init__(self, queue='cic-notify'):
@@ -22,17 +46,9 @@ class Api:
         :param queue: The queue on which to execute notification tasks
         :type queue: str
         """
-        registered_tasks = app.tasks
+        self.sms_tasks = get_sms_queue_tasks(app)
-        self.sms_tasks = []
+        logg.debug('sms tasks {}'.format(self.sms_tasks))
 
-        for task in registered_tasks.keys():
-            logg.debug(f'Found: {task} {registered_tasks[task]}')
-            match = re.match(sms_tasks_matcher, task)
-            if match:
-                self.sms_tasks.append(task)
-
-        self.queue = queue
-        logg.info(f'api using queue: {self.queue}')
 
     def sms(self, message, recipient):
         """This function chains all sms tasks in order to send a message, log and persist said data to disk
@@ -44,12 +60,17 @@ class Api:
         :rtype: Celery.Task
         """
         signatures = []
-        for task in self.sms_tasks:
+        for q in self.sms_tasks:
-            signature = celery.signature(task)
+            signature = celery.signature(
+                    q[1],
+                    [
+                        message,
+                        recipient,
+                        ],
+                    queue=q[0],
+                    )
             signatures.append(signature)
-            signature_group = celery.group(signatures)
+
-            result = signature_group.apply_async(
+        t = celery.group(signatures)()
-                args=[message, recipient],
+
-                queue=self.queue
+        return t
-            )
-            return result

apps/cic-notify/cic_notify/runnable/send.py

@@ -0,0 +1,76 @@
+# standard imports
+import sys
+import os
+import logging
+import argparse
+import tempfile
+
+# external imports
+import celery
+import confini
+
+# local imports
+from cic_notify.api import Api
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+config_dir = os.path.join('/usr/local/etc/cic-notify')
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-c', type=str, default=config_dir, help='config file')
+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('-v', action='store_true', help='be verbose')
+argparser.add_argument('-vv', action='store_true', help='be more verbose')
+argparser.add_argument('recipient', type=str, help='notification recipient')
+argparser.add_argument('message', type=str, help='message text')
+args = argparser.parse_args()
+
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
+config = confini.Config(args.c, args.env_prefix)
+config.process()
+config.censor('PASSWORD', 'DATABASE')
+config.add(args.recipient, '_RECIPIENT', True)
+config.add(args.message, '_MESSAGE', True)
+
+# set up celery
+app = celery.Celery(__name__)
+
+broker = config.get('CELERY_BROKER_URL')
+if broker[:4] == 'file':
+    bq = tempfile.mkdtemp()
+    bp = tempfile.mkdtemp()
+    app.conf.update({
+            'broker_url': broker,
+            'broker_transport_options': {
+                'data_folder_in': bq,
+                'data_folder_out': bq,
+                'data_folder_processed': bp,
+            },
+            },
+            )
+    logg.warning('celery broker dirs queue i/o {} processed {}, will NOT be deleted on shutdown'.format(bq, bp))
+else:
+    app.conf.update({
+        'broker_url': broker,
+        })
+
+result = config.get('CELERY_RESULT_URL')
+if result[:4] == 'file':
+    rq = tempfile.mkdtemp()
+    app.conf.update({
+        'result_backend': 'file://{}'.format(rq),
+        })
+    logg.warning('celery backend store dir {} created, will NOT be deleted on shutdown'.format(rq))
+else:
+    app.conf.update({
+        'result_backend': result,
+        })
+
+if __name__ == '__main__':
+    a = Api()
+    t = a.sms(config.get('_RECIPIENT'), config.get('_MESSAGE'))

apps/cic-notify/docker/Dockerfile

@@ -6,7 +6,7 @@ RUN apt-get update && \
 WORKDIR /usr/src/cic-notify
 
 ARG pip_extra_index_url_flag='--index https://pypi.org/simple --extra-index-url https://pip.grassrootseconomics.net:8433'
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a61
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a62
 
 COPY cic-notify/setup.cfg \
 	 cic-notify/setup.py \

apps/cic-notify/setup.cfg

@@ -45,3 +45,4 @@ testing =
 [options.entry_points]
 console_scripts =
 	cic-notify-tasker = cic_notify.runnable.tasker:main
+	cic-notify-send = cic_notify.runnable.send:main

apps/contract-migration/docker/Dockerfile

@@ -57,7 +57,7 @@ WORKDIR /home/grassroots
 USER grassroots
 
 ARG pip_extra_index_url=https://pip.grassrootseconomics.net:8433
-ARG cic_base_version=0.1.2a61
+ARG cic_base_version=0.1.2a62
 ARG cic_eth_version=0.11.0b1
 ARG sarafu_faucet_version=0.0.2a19
 ARG cic_contracts_version=0.0.2a2