Worker health check implementation

Jenkinsfile

@@ -4,5 +4,6 @@
 
 deliveryPipeline ([
 	buildContainer: 'plivo/jenkins-ci/python/2.7.14/ci-base/ubuntu/trusty:18.02.01.139',
-	disableQAStages: true
+	disableQAStages: true,
+	trivyBypass: true
 ])

docs/configuration.rst

@@ -119,3 +119,4 @@ A sample configuration file looks like this. You can also get this configuration
     ;; tcp connection settings
     port                 : 6379
     host                 : 127.0.0.1
+    worker_health_key    :  msg-worker-nps:us-east-1

requirements.txt

@@ -10,4 +10,4 @@ itsdangerous==0.24
 msgpack==0.5.6
 ujson==2.0.0
 uWSGI==2.0.21
-SharQ==1.3.0
+git+https://github.com/plivo/sharq.git@worker-status

sharq.conf

@@ -18,4 +18,5 @@ unix_socket_path          : /tmp/redis.sock
 ;; tcp connection settings
 port                      : 6379
 host                      : 127.0.0.1
-clustered                 : true  
+clustered                 : true
+worker_health_key         : msg-worker-nps:us-east-1  

sharq_server/server.py

@@ -56,6 +56,9 @@ def __init__(self, config_path):
         self.app.add_url_rule(
             '/deepstatus/',
             view_func=self._view_deep_status, methods=['GET'])
+        self.app.add_url_rule(
+            '/workerhealthstatus/',
+            view_func=self._view_worker_health_status, methods=['GET'])
 
     def requeue(self):
         """Loop endlessly and requeue expired jobs."""
@@ -263,6 +266,27 @@ def _view_deep_status(self):
             for line in traceback.format_exc().splitlines():
                 print(line)
             raise Exception
+    def _view_worker_health_status(self):
+        """Checks worker health status"""
+        try:
+            response = {}
+            key = self.config.get('redis', 'worker_health_key')
+            with self.sq.redis_client().lock('worker-health-lock-key', timeout=2): # will be changed later
+                value = self.sq.worker_health_status(key)
+                print(key, value)
+                if value is None:
+                    return jsonify(**response), 500
+                return jsonify(**response)
+        except LockError:
+            # the lock wasn't acquired within specified time
+            print("worker-health-lock-key lock not acquired", key)
+            return jsonify(**response)
+        except Exception as e:
+            print(e)
+            import traceback
+            for line in traceback.format_exc().splitlines():
+                print(line)
+            raise Exception
 
     def _view_clear_queue(self, queue_type, queue_id):
         """remove queue from SharQ based on the queue_type and queue_id."""

src/config/nginx-sharq.conf

@@ -15,6 +15,11 @@ server {
             uwsgi_pass unix:///var/run/sharq/sharq.sock;
             include uwsgi_params;
         }
+        location /workerhealthstatus/ {
+            log_not_found off;
+            uwsgi_pass unix:///var/run/sharq/sharq.sock;
+            include uwsgi_params;
+        }
 
         location / {
             # Not needed because it's all in the VPC

src/config/sharq.conf

@@ -11,4 +11,5 @@ unix_socket_path          = /tmp/redis.sock
 port                      = 6379
 host                      = 127.0.0.1
 clustered                 = false
-password                  = hello
+password                  = hello
+worker_health_key         = msg-worker-nps:us-east-1

src/config/sharq.conf.ctmpl

@@ -22,3 +22,4 @@ port                 : {{ printf "%s/%s/%s/%s/config/redis/port" $team $appenv $
 host                 : {{ printf "%s/%s/%s/%s/config/redis/host" $team $appenv $sharq_type $region | key }}
 clustered            : {{ printf "%s/%s/%s/%s/config/redis/clustered" $team $appenv $sharq_type $region | key }}
 password             : {{ printf "%s/%s/%s/%s/config/redis/password" $team $appenv $sharq_type $region | key }}
+worker_health_key    : {{ printf "%s/%s/%s/%s/config/redis/worker_health_key" $team $appenv $sharq_type $region | key }}