Merge branch '0.9.x' of github.com:pricingassistant/mrq into 0.9.x

* '0.9.x' of github.com:pricingassistant/mrq:
  update known queues when requeuing jobs
  fix makefile
  add minified js
  add external ip on dashboard
  add external ip in config
  don't search entire queue
  fix runtime error in get_worker_report
  Correctly patch HTTPConnection.request for Python 3.6 (#184)
  add task to repopulate known queues

Author: sylvinus

Makefile

@@ -44,6 +44,9 @@ clean:
 	find . -path ./venv -prune -o -name "*.pyc" -exec rm {} \;
 	find . -name __pycache__ | xargs rm -r
 
+build_dashboard:
+	cd mrq/dashboard/static && npm install && mkdir -p bin && npm run build
+
 dashboard:
 	python mrq/dashboard/app.py
 

mrq/basetasks/cleaning.py

@@ -88,6 +88,19 @@ def run(self, params):
             Queue(str(queue)).add_to_known_queues()
 
 
+class PopulateKnownQueues(Task):
+
+    """
+        Populates the known queues in Redis.
+    """
+
+    def run(self, params):
+
+        queues = Queue.all()
+        for queue in queues:
+            Queue(queue, add_to_known_queues=True)
+
+
 class CleanKnownQueues(Task):
 
     """
@@ -123,8 +136,9 @@ def run(self, params):
                 q = Queue(queue, add_to_known_queues=False)
                 size = q.size()
                 if check_mongo:
-                    size += connections.mongodb_jobs.mrq_jobs.count({"queue": queue})
-                if size == 0:
+                    has_job = connections.mongodb_jobs.mrq_jobs.find_one({"queue": queue})
+
+                if size == 0 or has_job is None:
                     removed_queues.append(queue)
                     print("Removing empty queue '%s' from known queues ..." % queue)
                     if not pretend:

mrq/basetasks/utils.py

@@ -134,6 +134,12 @@ def perform_action(self, action, query, destination_queue):
                         "_id": {"$in": jobs_by_queue[queue]}
                     }, {"$set": updates}, multi=True)
 
+                if destination_queue is None:
+                    Queue.ensure_known_queues(jobs_by_queue.iterkeys())
+
+            if destination_queue is not None:
+                Queue.ensure_known_queues([destination_queue])
+
         print(stats)
 
         return stats

mrq/config.py

@@ -454,6 +454,13 @@ def add_parser_args(parser, config_type):
             type=str,
             help='Overwrite the local IP, to be displayed in the dashboard.')
 
+        parser.add_argument(
+            '--external_ip',
+            default=None,
+            action="store",
+            type=str,
+            help='Overwrite the external IP, to be displayed in the dashboard.')
+
         parser.add_argument(
             '--max_latency',
             default=1.,

mrq/dashboard/static/bin/0.bundle.js

@@ -58,7 +58,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models", "moment
             "sType":"string",
             "sWidth":"150px",
             "mData":function(source, type/*, val*/) {
-              return "<a href='/#jobs?worker="+source._id+"'>"+source.name+"</a><br/><small>"+source.config.local_ip + " " + source._id+"</small>";
+              return "<a href='/#jobs?worker="+source._id+"'>"+source.name+"</a><br/><small>"+ (source.config.external_ip || source.config.local_ip) + " " + source._id+"</small>";
             }
           },
           {

mrq/dashboard/static/js/views/workers.js

@@ -2,8 +2,7 @@
   "name": "mrq-dashboard",
   "version": "0.9.1",
   "description": "PricingAssistant MRQ dashboard",
-  "dependencies": {
-  },
+  "dependencies": {},
   "engines": {
     "node": "^7.5.0"
   },

mrq/dashboard/static/package.json

@@ -304,7 +304,7 @@ def makefile(self, *args, **kwargs):
             newsock = self._obj.makefile(*args, **kwargs)
             return mrq_wrapped_socket(newsock, self._parent_connection)
 
-    def request(old_method, self, method, url, body=None, headers=None):
+    def request(old_method, self, method, url, body=None, headers=None, *args, **kwargs):
 
         if headers is None:
             headers = {}

mrq/monkey.py

@@ -10,6 +10,7 @@
 
 import sys
 from future import standard_library
+from itertools import chain
 
 PY3 = sys.version_info > (3,)
 standard_library.install_aliases()
@@ -119,6 +120,13 @@ def get_retry_queue(self):
         """ Return the name of the queue where retried jobs will be queued """
         return self.id
 
+    @classmethod
+    def ensure_known_queues(cls, queues):
+        """ List all previously known queues """
+        now = time.time()
+        params = chain.from_iterable((now, queue) for queue in queues)
+        context.connections.redis.zadd(Queue.redis_key_known_queues(), *params)
+
     def add_to_known_queues(self, timestamp=None):
         """ Adds this queue to the shared list of known queues """
         now = timestamp or time.time()

mrq/queue.py

@@ -208,7 +208,7 @@ def get_worker_report(self, with_memory=False):
             its jobs. """
 
         greenlets = []
-        for greenlet in self.gevent_pool:
+        for greenlet in list(self.gevent_pool):
             g = {}
             short_stack = []
             stack = traceback.format_stack(greenlet.gr_frame)
@@ -263,6 +263,7 @@ def get_worker_report(self, with_memory=False):
             "scheduler",
             "name",
             "local_ip",
+            "external_ip",
             "agent_id",
             "worker_group",
             "worker_profile"