Add health check to cic-eth-tasker

.gitignore

@@ -3,4 +3,8 @@ service-configs/*
 node_modules
 __pycache__
 *.pyc
-*.o
+*.o
+gmon.out
+*.egg-info
+dist/
+build/

apps/cic-eth/cic_eth/k8s/db.py

@@ -0,0 +1,7 @@
+from cic_eth.db.models.base import SessionBase
+
+def health():
+    session = SessionBase.create_session()
+    session.execute('SELECT count(*) from alembic_version')
+    session.close()
+    return True

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

@@ -15,6 +15,7 @@ from chainlib.connection import RPCConnection
 from chainlib.eth.connection import EthUnixSignerConnection
 from chainlib.chain import ChainSpec
 from chainqueue.db.models.otx import Otx
+import liveness.linux
 
 # local imports
 from cic_eth.eth import (
@@ -52,6 +53,7 @@ from cic_eth.registry import (
         connect_token_registry,
         )
 
+
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
 
@@ -90,14 +92,15 @@ config.censor('PASSWORD', 'DATABASE')
 config.censor('PASSWORD', 'SSL')
 logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
+health_modules = config.get('CIC_HEALTH_MODULES', [])
+if len(health_modules) != 0:
+    health_modules = health_modules.split(',')
+logg.debug('health mods {}'.format(health_modules))
+
 # connect to database
 dsn = dsn_from_config(config)
 SessionBase.connect(dsn, pool_size=int(config.get('DATABASE_POOL_SIZE')), debug=config.true('DATABASE_DEBUG'))
 
-# verify database connection with minimal sanity query
-session = SessionBase.create_session()
-session.execute('select version_num from alembic_version')
-session.close()
 
 # set up celery
 current_app = celery.Celery(__name__)
@@ -139,6 +142,7 @@ RPCConnection.register_location(config.get('SIGNER_SOCKET_PATH'), chain_spec, 's
 
 Otx.tracing = config.true('TASKS_TRACE_QUEUE_STATUS')
 
+liveness.linux.load('cic-eth', health_modules)
 
 def main():
     argv = ['worker']
@@ -173,8 +177,10 @@ def main():
         logg.info('using trusted address {}'.format(address))
     connect_declarator(rpc, chain_spec, trusted_addresses)
     connect_token_registry(rpc, chain_spec)
-    
+   
+    liveness.linux.set('cic-eth')
     current_app.worker_main(argv)
+    liveness.linux.reset('cic-eth')
 
 
 @celery.signals.eventlet_pool_postshutdown.connect

apps/cic-eth/config/cic.ini

@@ -3,3 +3,4 @@ registry_address =
 chain_spec = evm:bloxberg:8996
 tx_retry_delay = 
 trust_address =
+health_modules = cic_eth.k8s.db

apps/cic-eth/requirements.txt

@@ -1,4 +1,4 @@
-cic-base~=0.1.2a76
+cic-base==0.1.2a79+build.5c9102de
 celery==4.4.7
 crypto-dev-signer~=0.4.14b2
 confini~=0.3.6rc3

apps/cic-eth/setup.cfg

@@ -38,6 +38,7 @@ packages =
 	cic_eth.runnable.daemons.filters
 	cic_eth.callbacks
 	cic_eth.sync
+	cic_eth.k8s
 scripts =
 	./scripts/migrate.py
 

apps/util/liveness/MANIFEST.in

@@ -0,0 +1 @@
+include *health*.sh

apps/util/liveness/liveness/linux.py

@@ -14,14 +14,15 @@ def load(namespace, check_strs, rundir='/run'):
 
     checks = []
     for m in check_strs:
-        logg.debug('added liveness check module {}'.format(str(m)))
+        logg.debug('added liveness check: {}'.format(str(m)))
         module = importlib.import_module(m)
         checks.append(module)
 
     for check in checks:
         r = check.health()
         if r == False:
-            raise RuntimeError('check {} failed'.format(str(check)))
+            raise RuntimeError('liveness check {} failed'.format(str(check)))
+        logg.info('liveness check passed: {}'.format(str(check)))
 
     app_rundir = os.path.join(rundir, namespace)
     os.makedirs(app_rundir) # should not already exist

apps/util/liveness/setup.py

@@ -0,0 +1,7 @@
+from setuptools import setup
+setup(
+        name='liveness',
+        version='0.0.1a2',
+        packages=['liveness'],
+        include_package_data=True,
+        )