Merge branch 'master' of github.com:eventmq/eventmq

11 files changed, 134 insertions, 49 deletions

diff --git a/bin/send_msg b/bin/send_msg
index c3e88fa..33cfb1a 100755
--- a/bin/send_msg
+++ b/bin/send_msg
@@ -22,7 +22,7 @@ if __name__ == "__main__":
             'callable': 'work_job',
             'class_args': ('blurp',),
             'class_kwargs': {'kwarg1': True},
-            'args': (10, ),
+            'args': (1, ),
             'kwargs': {}
             }]
 
diff --git a/docs/settings_file.rst b/docs/settings_file.rst
index 43c922c..c758a16 100644
--- a/docs/settings_file.rst
+++ b/docs/settings_file.rst
@@ -30,6 +30,20 @@ Default: 'tcp://127.0.0.1:47291'
 
 The address used to listen for connections from workers
 
+wal
+=======
+Default: '/var/log/eventmq/wal.log'
+
+Write-ahead Log for replaying messages received by the Router.  Will
+try to create the directory specified and append to the filename given.
+Requires correct permissions to write to the given file.
+
+wal_enabled
+===============
+Default: False
+
+Enable or disable the Write-ahead Log
+
 *********
 Scheduler
 *********
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index 60bb61c..a02a091 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3.4.6'
+__version__ = '0.3.4.9'
 
 PROTOCOL_VERSION = 'eMQP/1.0'
 
diff --git a/eventmq/conf.py b/eventmq/conf.py
index 4749e6b..8609716 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -97,4 +97,7 @@ SETUP_CALLABLE = ''
 KILL_GRACE_PERIOD = 300
 GLOBAL_TIMEOUT = 300
 
+WAL = '/var/log/eventmq/wal.log'
+WAL_ENABLED = False
+
 # }}}
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 4840fac..c900495 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -146,8 +146,6 @@ class JobManager(HeartbeatMixin, EMQPService):
         Starts the actual event loop. Usually called by :meth:`start`
         """
         # Acknowledgment has come
-        # Send a READY for each available worker
-
         self.status = STATUS.running
 
         try:
@@ -186,9 +184,9 @@ class JobManager(HeartbeatMixin, EMQPService):
                     else:
                         try:
                             events = self.poller.poll(1000)
-                        except zmq.ZMQError:
+                        except zmq.ZMQError as e:
                             logger.debug('Disconnecting due to ZMQError while'
-                                         ' polling')
+                                         ' polling: {}'.format(e))
                             sendmsg(self.outgoing, KBYE)
                             self.received_disconnect = True
                             continue
@@ -212,6 +210,15 @@ class JobManager(HeartbeatMixin, EMQPService):
         except Exception:
             logger.exception("Unhandled exception in main jobmanager loop")
 
+        # Cleanup
+        if hasattr(self, '_workers'):
+            del self._workers
+
+        # Flush the queues with workers
+        self.request_queue = mp_queue()
+        self.finished_queue = mp_queue()
+        logger.info("Reached end of event loop")
+
     def handle_response(self, resp):
         """
         Handles a response from a worker process to the jobmanager
@@ -240,7 +247,8 @@ class JobManager(HeartbeatMixin, EMQPService):
 
         """
 
-        logger.debug(resp)
+        if conf.SUPER_DEBUG:
+            logger.debug(resp)
         pid = resp['pid']
         msgid = resp['msgid']
         callback = resp['callback']
@@ -383,9 +391,6 @@ class JobManager(HeartbeatMixin, EMQPService):
         necessary
         """
         # Kill workers that aren't alive
-        logger.debug("Jobs in flight: {}".format(len(self.jobs_in_flight)))
-        logger.debug("Total requests: {}".format(self.total_requests))
-        logger.debug("Total ready sent: {}".format(self.total_ready_sent))
         try:
             [self.kill_worker(w.pid, signal.SIGKILL) for w in self.workers
              if not w.is_alive]
diff --git a/eventmq/log.py b/eventmq/log.py
index 5564094..3d62165 100644
--- a/eventmq/log.py
+++ b/eventmq/log.py
@@ -17,7 +17,10 @@ log module for eventmq
 
 this needs so much work.
 """
+import errno
 import logging
+import os
+import time
 
 import zmq
 import zmq.log.handlers
@@ -46,6 +49,7 @@ class handlers(object):
     """
     PUBLISH_HANDLER = PUBHandler
     STREAM_HANDLER = logging.StreamHandler
+    FILE_HANDLER = logging.FileHandler
 
 
 def setup_logger(base_name, formatter=FORMAT_STANDARD,
@@ -75,3 +79,30 @@ def setup_logger(base_name, formatter=FORMAT_STANDARD,
     logger.addHandler(handler)
 
     return logger
+
+
+def setup_wal_logger(base_name, filename, handler=handlers.FILE_HANDLER):
+    """
+    Write-ahead Log for replaying messages.  Should only contain
+    commands on the data path (REQUEST, SCHEDULE, UNSCHEDULE)
+    """
+
+    if not os.path.exists(os.path.dirname(filename)):
+        try:
+            os.makedirs(os.path.dirname(filename))
+        except OSError as exc:  # Guard against race condition
+            if exc.errno != errno.EEXIST:
+                raise
+
+    with open(filename, "a+") as f:
+        f.close()
+
+    wal = logging.getLogger(base_name)
+    wal_handler = handler(filename)
+    formatter = logging.Formatter('%(asctime)s %(message)s')
+    formatter.converter = time.gmtime
+    wal_handler.setFormatter(formatter)
+    wal.addHandler(wal_handler)
+    wal.setLevel(logging.INFO)
+
+    return wal
diff --git a/eventmq/router.py b/eventmq/router.py
index 2943e3c..7c221bc 100644
--- a/eventmq/router.py
+++ b/eventmq/router.py
@@ -22,7 +22,7 @@ import json  # deserialize queues in on_inform. should be refactored
 import logging
 import signal
 
-from eventmq.log import setup_logger
+from eventmq.log import setup_logger, setup_wal_logger
 from . import conf, constants, exceptions, poller, receiver
 from .constants import (
     CLIENT_TYPE, DISCONNECT, KBYE, PROTOCOL_VERSION, ROUTER_SHOW_SCHEDULERS,
@@ -41,6 +41,7 @@ from .utils.timeutils import monotonic, timestamp
 
 
 logger = logging.getLogger(__name__)
+wal_logger = logging.getLogger('eventmq-wal')
 
 
 class Router(HeartbeatMixin):
@@ -94,6 +95,13 @@ class Router(HeartbeatMixin):
         #: workers available to take the job
         self.waiting_messages = {}
 
+        # Key: Queue.name, Value: # of messages sent to workers on that queue
+        # Includes REQUESTS in flight but not REQUESTS queued
+        self.processed_message_counts = {}
+
+        # Same as above but Key: Worker.uuid
+        self.processed_message_counts_by_worker = {}
+
         #: Tracks the last time the scheduler queue was cleaned out of dead
         #: schedulers
         self._meta['last_scheduler_cleanup'] = 0
@@ -118,8 +126,6 @@ class Router(HeartbeatMixin):
         #: Excecuted function tracking dictionary
         #: Key: msgid of msg each REQUEST received and forwarded to a worker
         #: Value: (function_name, queue_name)
-        self.executed_functions = {}
-
         #: Set to True when the router should die.
         self.received_disconnect = False
 
@@ -171,6 +177,7 @@ class Router(HeartbeatMixin):
 
             if events.get(self.incoming) == poller.POLLIN:
                 msg = self.incoming.recv_multipart()
+                self.handle_wal_log(msg)
                 self.process_client_message(msg)
 
             if events.get(self.outgoing) == poller.POLLIN:
@@ -179,7 +186,8 @@ class Router(HeartbeatMixin):
 
             if events.get(self.administrative_socket) == poller.POLLIN:
                 msg = self.administrative_socket.recv_multipart()
-                logger.debug('ADMIN: {}'.format(msg))
+                if conf.SUPER_DEBUG:
+                    logger.debug('ADMIN: {}'.format(msg))
                 # ##############
                 # Admin Commands
                 # ##############
@@ -331,8 +339,9 @@ class Router(HeartbeatMixin):
         """
 
         orig_msgid = msg[1]
-        logger.info('Received REPLY from {} (msgid: {}, ACK msgid: {})'.format(
-            sender, msgid, orig_msgid))
+        if conf.SUPER_DEBUG:
+            logger.debug('Received REPLY from {} (msgid: {}, ACK msgid: {})'.
+                         format(sender, msgid, orig_msgid))
 
         if orig_msgid in self.job_latencies:
             elapsed_secs = (monotonic()
@@ -484,16 +493,20 @@ class Router(HeartbeatMixin):
 
         try:
             # Check if msg type is for executing function
-            if 'run' in msg and len(msg) > 2:
-                args_list = json.loads(msg[2])
-                args_dict = args_list[1]
-                function = args_dict.get('callable')
-                if function:
-                    self.executed_functions[msgid] = (function, queue_name)
             self.job_latencies[msgid] = (monotonic(), queue_name)
 
             # Rebuild the message to be sent to the worker. fwdmsg will
             # properly address the message.
+            if queue_name not in self.processed_message_counts:
+                self.processed_message_counts[queue_name] = 1
+            else:
+                self.processed_message_counts[queue_name] += 1
+
+            if queue_name not in self.processed_message_counts_by_worker:
+                self.processed_message_counts_by_worker[worker_addr] = 1
+            else:
+                self.processed_message_counts_by_worker[worker_addr] += 1
+
             fwdmsg(self.outgoing, worker_addr, ['', constants.PROTOCOL_VERSION,
                                                 'REQUEST', msgid, ] + msg)
 
@@ -694,6 +707,21 @@ class Router(HeartbeatMixin):
         """
         self.workers[worker_id]['available_slots'] += 1
 
+    def handle_wal_log(self, original_msg):
+
+        try:
+            message = parse_router_message(original_msg)
+        except exceptions.InvalidMessageError:
+            logger.exception('Invalid message from clients: {}'.format(
+                str(original_msg)))
+            return
+
+        command = message[1]
+
+        if conf.WAL_ENABLED and \
+           command in ("REQUEST", "SCHEDULE", "UNSCHEDULE"):
+            wal_logger.info(original_msg)
+
     def process_client_message(self, original_msg, depth=0):
         """
         Args:
@@ -876,8 +904,10 @@ class Router(HeartbeatMixin):
            (str) Serialized information about the current state of the router.
         """
         return json.dumps({
-            'job_latencies': self.job_latencies,
-            'executed_functions': self.executed_functions,
+            '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,
@@ -915,6 +945,7 @@ class Router(HeartbeatMixin):
         """
         setup_logger('eventmq')
         import_settings()
+        setup_wal_logger('eventmq-wal', conf.WAL)
         self.start(frontend_addr=conf.FRONTEND_ADDR,
                    backend_addr=conf.BACKEND_ADDR,
                    administrative_addr=conf.ADMINISTRATIVE_ADDR)
diff --git a/eventmq/tests/test_router.py b/eventmq/tests/test_router.py
index fdb02d0..e9efa25 100644
--- a/eventmq/tests/test_router.py
+++ b/eventmq/tests/test_router.py
@@ -776,8 +776,9 @@ class TestCase(unittest.TestCase):
         # correctly and what not.
         self.assertEqual(
             json.loads(json.dumps({
-                'job_latencies': self.router.job_latencies,
-                'executed_functions': self.router.executed_functions,
+                'job_latencies_count': len(self.router.job_latencies),
+                'processed_messages': {},
+                'processed_messages_by_worker': {},
                 'waiting_message_counts': [
                     '{}: {}'.format(
                         q,
diff --git a/eventmq/utils/classes.py b/eventmq/utils/classes.py
index 39c600b..73c6364 100644
--- a/eventmq/utils/classes.py
+++ b/eventmq/utils/classes.py
@@ -345,10 +345,9 @@ class ZMQReceiveMixin(object):
         Receive a message
         """
         msg = self.zsocket.recv()
-        if conf.SUPER_DEBUG:
-            if not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
-               not conf.HIDE_HEARTBEAT_LOGS:
-                logger.debug('Received message: {}'.format(msg))
+        if not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
+                not conf.HIDE_HEARTBEAT_LOGS:
+            logger.debug('Received message: {}'.format(msg))
         return msg
 
     def recv_multipart(self):
@@ -356,13 +355,12 @@ class ZMQReceiveMixin(object):
         Receive a multipart message
         """
         msg = self.zsocket.recv_multipart()
-        if conf.SUPER_DEBUG:
-            # If it's not at least 4 frames long then most likely it isn't an
-            # eventmq message
-            if len(msg) >= 4 and \
-               not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
-               not conf.HIDE_HEARTBEAT_LOGS:
-                logger.debug('Received message: {}'.format(msg))
+        # If it's not at least 4 frames long then most likely it isn't an
+        # eventmq message
+        if len(msg) >= 4 and \
+            not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
+                not conf.HIDE_HEARTBEAT_LOGS:
+            logger.debug('Received message: {}'.format(msg))
         return msg
 
 
@@ -402,13 +400,12 @@ class ZMQSendMixin(object):
 
         msg = encodify(headers + message)
 
-        if conf.SUPER_DEBUG:
-            # If it's not at least 4 frames long then most likely it isn't an
-            # eventmq message
-            if len(msg) > 4 and \
-               not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
-               not conf.HIDE_HEARTBEAT_LOGS:
-                logger.debug('Sending message: %s' % str(msg))
+        # If it's not at least 4 frames long then most likely it isn't an
+        # eventmq message
+        if len(msg) > 4 and \
+            not ("HEARTBEAT" == msg[2] or "HEARTBEAT" == msg[3]) or \
+                not conf.HIDE_HEARTBEAT_LOGS:
+            logger.debug('Sending message: %s' % str(msg))
 
         try:
             self.zsocket.send_multipart(msg,
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 859b831..1621578 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -109,16 +109,19 @@ class MultiprocessWorker(Process):
                 timeout = payload.get("timeout") or conf.GLOBAL_TIMEOUT
                 msgid = payload.get('msgid', '')
                 callback = payload.get('callback', '')
-                logger.debug("Putting on thread queue msgid: {}".format(
-                    msgid))
+
+                if conf.SUPER_DEBUG:
+                    logger.debug("Putting on thread queue msgid: {}".format(
+                        msgid))
 
                 worker_queue.put(payload['params'])
 
                 try:
                     return_val = worker_result_queue.get(timeout=timeout)
 
-                    logger.debug("Got from result queue msgid: {}".format(
-                        msgid))
+                    if conf.SUPER_DEBUG:
+                        logger.debug("Got from result queue msgid: {}".format(
+                            msgid))
                 except Queue.Empty:
                     return_val = 'TimeoutError'
 
diff --git a/setup.py b/setup.py
index 9ac6089..553c86b 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ from setuptools import find_packages, setup
 
 setup(
     name='eventmq',
-    version='0.3.4.6',
+    version='0.3.4.9',
     description='EventMQ job execution and messaging system based on ZeroMQ',
     packages=find_packages(),
     install_requires=['pyzmq==15.4.0',