Merge pull request #17 from plivo/sharq-ft

multiregion sharq config

.gitignore

@@ -57,3 +57,4 @@ Dockerfile_supervisor
 test_sharq.py
 .idea/*
 venv/*
+.vscode/

requirements.txt

@@ -1,14 +1,13 @@
 Flask==0.10.1
 Jinja2==2.7.2
 MarkupSafe==0.23
-SharQ==0.4.0
 Werkzeug==0.9.4
 argparse==1.2.1
 gevent==1.0.1
 greenlet==0.4.2
 gunicorn==19.0.0
 itsdangerous==0.24
 msgpack-python==0.4.2
-redis==2.10.1
 ujson==1.33
 wsgiref==0.1.2
+SharQ==0.5.2

setup.py

@@ -14,7 +14,6 @@
     packages=['sharq_server'],
     py_modules=['runner'],
     install_requires=[
-        'SharQ==0.4.0',
         'Flask==0.10.1',
         'Jinja2==2.7.2',
         'MarkupSafe==0.23',

sharq.conf

@@ -18,3 +18,4 @@ unix_socket_path          : /tmp/redis.sock
 ;; tcp connection settings
 port                      : 6379
 host                      : 127.0.0.1
+clustered                 : true  

sharq_server/server.py

@@ -53,6 +53,9 @@ def __init__(self, config_path):
         self.app.add_url_rule(
             '/deletequeue/<queue_type>/<queue_id>/',
             view_func=self._view_clear_queue, methods=['DELETE'])
+        self.app.add_url_rule(
+            '/deepstatus/',
+            view_func=self._view_deep_status, methods=['GET'])
 
     def requeue(self):
         """Loop endlessly and requeue expired jobs."""
@@ -104,6 +107,10 @@ def _view_dequeue(self, queue_type):
             if response['status'] == 'failure':
                 return jsonify(**response), 404
         except Exception, e:
+            print e
+            import traceback
+            for line in traceback.format_exc().splitlines():
+                print line
             response['message'] = e.message
             return jsonify(**response), 400
 
@@ -176,9 +183,17 @@ def _view_metrics(self, queue_type, queue_id):
             return jsonify(**response), 400
 
         return jsonify(**response)
-
+
+    def _view_deep_status(self):
+        """Checks  underlying data store health"""
+        self.sq.ping()
+        response = {
+            'status': "success"
+        }
+        return jsonify(**response)
+
     def _view_clear_queue(self, queue_type, queue_id):
-        """remove queueu from SharQ based on the queue_type and queue_id."""
+        """remove queue from SharQ based on the queue_type and queue_id."""
         response = {
             'status': 'failure'
         }

src/config/nginx-sharq.conf

@@ -10,6 +10,12 @@ server {
             include uwsgi_params;
         }
 
+        location /deepstatus/ {
+            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
             log_not_found off;

src/config/sharq.conf.ctmpl

@@ -1,3 +1,7 @@
+{{$region := env "REGION"}}
+{{$appenv := env "ENVIRONMENT"}}
+{{$team := env "TEAM"}}
+{{$sharq_type := env "SHARQ_TYPE"}}
 [sharq]
 job_expire_interval  : 45000 ; in milliseconds
 job_requeue_interval : 1000 ; in milliseconds
@@ -11,11 +15,12 @@ accesslog            : /tmp/sharq.log ; optional
 
 [redis]
 db                   : 0
-key_prefix           : sharq_server
+key_prefix           : {{ printf "%s/%s/%s/%s/config/redis/key_prefix" $team $appenv $sharq_type $region | key }}
 conn_type            : tcp_sock ; or tcp_sock
 ;; unix connection settings
 ;; unix_socket_path     : /var/run/redis/redis.sock
 ;; tcp connection settings
 
-port                 : {{ printf "%s/%s/%s/config/redis/port" (env "TEAM") (env "ENVIRONMENT") (env "SHARQ_TYPE") | key | parseInt }}
-host                 : {{ printf "%s/%s/%s/config/redis/host" (env "TEAM") (env "ENVIRONMENT") (env "SHARQ_TYPE") | key }}
+port                 : {{ printf "%s/%s/%s/%s/config/redis/port" $team $appenv $sharq_type $region | key | parseInt }}
+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 }}