Merge branch 'master' into dependabot/pip/psutil-5.6.6dependabot/pip/psutil-5.6.6

2 files changed, 64 insertions, 25 deletions

diff --git a/eventmq/router.py b/eventmq/router.py
index 3c141c3..49ffb3a 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -51,6 +51,7 @@ class Router(HeartbeatMixin):
     """
     A simple router of messages
     """
+
     def __init__(self, *args, **kwargs):
         super(Router, self).__init__(*args, **kwargs)  # Creates _meta
 
@@ -532,7 +533,7 @@ class Router(HeartbeatMixin):
             # Recursively try again. TODO: are there better options?
             self.process_client_message(
                 [sender, '', PROTOCOL_VERSION, 'REQUEST', msgid] + msg,
-                depth=depth+1)
+                depth=depth + 1)
 
     def clean_up_dead_workers(self):
         """
@@ -601,7 +602,7 @@ class Router(HeartbeatMixin):
             self.queues[q[1]] = self.prioritize_queue_list(self.queues[q[1]])
 
         logger.debug('Added worker {} to the queues {}'.format(
-                 worker_id, queues))
+            worker_id, queues))
 
     def get_available_worker(self, queue_name=conf.DEFAULT_QUEUE_NAME):
         """
@@ -799,7 +800,7 @@ class Router(HeartbeatMixin):
             except exceptions.PeerGoneAwayError:
                 logger.debug("Scheduler {} has unexpectedly gone away. Trying "
                              "another scheduler.".format(scheduler_addr))
-                self.process_client_message(original_msg[1:], depth+1)
+                self.process_client_message(original_msg[1:], depth + 1)
 
         elif command == "UNSCHEDULE":
             # Forward the unschedule message to all schedulers
@@ -817,7 +818,7 @@ class Router(HeartbeatMixin):
                     logger.debug("Scheduler {} has unexpectedly gone away."
                                  " Schedule may still exist.".
                                  format(scheduler_addr))
-                    self.process_client_message(original_msg[1:], depth+1)
+                    self.process_client_message(original_msg[1:], depth + 1)
 
         elif command == DISCONNECT:
             self.on_disconnect(msgid, msg)
@@ -909,16 +910,49 @@ class Router(HeartbeatMixin):
         Return
            (str) Serialized information about the current state of the router.
         """
+        queue_latency_list = {}
+        queue_latency_count = {}
+        queue_max_latency_list = {}
+        queue_waiting_list = {}
+
+        now = monotonic()
+
+        for job in self.job_latencies:
+            queue = self.job_latencies[job][1]
+            latency = self.job_latencies[job][0]
+
+            if queue not in queue_latency_list:
+                queue_latency_list[queue] = latency
+                queue_latency_count[queue] = 1
+            else:
+                queue_latency_list[queue] += latency
+                queue_latency_count[queue] += 1
+
+            if queue not in queue_max_latency_list:
+                queue_max_latency_list[queue] = latency
+            else:
+                if queue_max_latency_list[queue] > latency:
+                    queue_max_latency_list[queue] = latency
+
+        for queue in queue_latency_list:
+            queue_latency_list[queue] = int(
+                (now - (queue_latency_list[queue] /
+                        max(queue_latency_count[queue], 1))) * 1000)
+
+        for queue in queue_max_latency_list:
+            queue_max_latency_list[queue] = int(
+                (now - queue_max_latency_list[queue]) * 1000)
+
+        for queue in self.waiting_messages:
+            queue_waiting_list[queue] = len(self.waiting_messages[queue])
+
         return json.dumps({
-            'job_latencies_count': len(self.job_latencies),
-            'processed_messages': self.processed_message_counts,
-            'processed_messages_by_worker':
-                self.processed_message_counts_by_worker,
-            'waiting_message_counts': [
-                '{}: {}'.
-                format(q,
-                       len(self.waiting_messages[q]))
-                for q in self.waiting_messages]
+            'inflight_messages_by_queue': queue_latency_count,
+            'latency_messages_by_queue': queue_latency_list,
+            'max_latency_messages_by_queue': queue_max_latency_list,
+            'waiting_messages_by_queue': queue_waiting_list,
+            'processed_messages_by_queue': self.processed_message_counts,
+            'processed_messages_by_worker': self.processed_message_counts_by_worker  # noqa
         })
 
     def get_workers_status(self):
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index 0a315e3..73612e7 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -464,6 +464,8 @@ class TestCase(unittest.TestCase):
         queue2_id = 'jimjam'
         nonexistent_queue1 = 'nonexistent'
 
+        t = monotonic()
+
         # To ensure the value was changed later because monotonic() is hard to
         # mock
         self.assertEqual(self.router._meta['last_worker_cleanup'], 0)
@@ -479,20 +481,20 @@ class TestCase(unittest.TestCase):
             # 3 in the future
             worker1_id: {
                 'queues': [(10, queue1_id), ],
-                'hb': monotonic() + 3,
+                'hb': t + 3,
                 'available_slots': 0,
             },
             # below the timeout
 
             worker2_id: {
                 'queues': [(10, queue2_id), (0, queue1_id)],
-                'hb': 0,
+                'hb': t - 2,
                 'available_slots': 2,
             },
             # below the timeout and a queue missing from self.router.queues
             worker3_id: {
                 'queues': [(10, queue2_id), (3, nonexistent_queue1)],
-                'hb': 0,
+                'hb': t - 2,
                 'available_slots': 0,
             },
         }
@@ -778,16 +780,19 @@ class TestCase(unittest.TestCase):
 
         # hacky, but the serialize/deserialize converts the keys to unicode
         # correctly and what not.
+        expected_object = {
+            'inflight_messages_by_queue': {},
+            'latency_messages_by_queue': {},
+            'max_latency_messages_by_queue': {},
+            'processed_messages_by_queue': {},
+            'processed_messages_by_worker': {},
+            'waiting_messages_by_queue': {
+                q: len(self.router.waiting_messages[q])
+                for q in self.router.waiting_messages
+            }
+        }
         self.assertEqual(
-            json.loads(json.dumps({
-                'job_latencies_count': len(self.router.job_latencies),
-                'processed_messages': {},
-                'processed_messages_by_worker': {},
-                'waiting_message_counts': [
-                    '{}: {}'.format(
-                        q,
-                        len(self.router.waiting_messages[q])) for q in self.router.waiting_messages]  # noqa
-            })),
+            json.loads(json.dumps(expected_object)),
             json.loads(self.router.get_status()))
 
         self.assertEqual(