More debugging logging

celery/bin/celeryd.py

@@ -289,4 +289,4 @@ def parse_options(arguments):
 
 if __name__ == "__main__":
     options = parse_options(sys.argv[1:])
-    run_worker(**options)
+    run_worker(**vars(options))

celery/pool.py

@@ -67,6 +67,8 @@ class TaskPool(object):
         on_return = curry(self.on_return, tid, callbacks, errbacks,
                           on_ack, meta)
 
+        self.logger.debug("TaskPool: Apply %s (args:%s kwargs:%s)" % (
+            target, args, kwargs))
         result = self._pool.apply_async(target, args, kwargs,
                                         callback=on_return)
 

celery/tests/test_worker_controllers.py

@@ -8,6 +8,14 @@ from celery.worker.controllers import Mediator, PeriodicWorkController
 from celery.worker.controllers import InfinityThread
 
 
+class MockTask(object):
+    task_id = 1234
+    task_name = "mocktask"
+
+    def __init__(self, value, **kwargs):
+        self.value = value
+
+
 class MyInfinityThread(InfinityThread):
 
     def on_iteration(self):
@@ -54,10 +62,10 @@ class TestMediator(unittest.TestCase):
         got = {}
 
         def mycallback(value):
-            got["value"] = value
+            got["value"] = value.value
 
         m = Mediator(bucket_queue, mycallback)
-        bucket_queue.put("George Constanza")
+        bucket_queue.put(MockTask("George Constanza"))
 
         m.on_iteration()
 
@@ -73,16 +81,19 @@ class TestPeriodicWorkController(unittest.TestCase):
 
         m.process_hold_queue()
 
-        hold_queue.put(("task1", datetime.now() - timedelta(days=1)))
+        hold_queue.put((MockTask("task1"),
+                        datetime.now() - timedelta(days=1)))
 
         m.process_hold_queue()
         self.assertRaises(Empty, hold_queue.get_nowait)
-        self.assertEquals(bucket_queue.get_nowait(), "task1")
+        self.assertEquals(bucket_queue.get_nowait().value, "task1")
         tomorrow = datetime.now() + timedelta(days=1)
-        hold_queue.put(("task2", tomorrow))
+        hold_queue.put((MockTask("task2"), tomorrow))
         m.process_hold_queue()
         self.assertRaises(Empty, bucket_queue.get_nowait)
-        self.assertEquals(hold_queue.get_nowait(), ("task2", tomorrow))
+        value, eta = hold_queue.get_nowait()
+        self.assertEquals(value.value, "task2")
+        self.assertEquals(eta, tomorrow)
 
     def test_run_periodic_tasks(self):
         bucket_queue = Queue()

celery/worker/controllers.py

@@ -6,6 +6,7 @@ Worker Controller Threads
 from celery.backends import default_periodic_status_backend
 from Queue import Empty as QueueEmpty
 from datetime import datetime
+from multiprocessing import get_logger
 import threading
 import time
 
@@ -70,12 +71,17 @@ class Mediator(InfinityThread):
         self.callback = callback
 
     def on_iteration(self):
+        logger = get_logger()
         try:
+            logger.debug("Mediator: Trying to get message from bucket_queue")
             # This blocks until there's a message in the queue.
             task = self.bucket_queue.get(timeout=1)
         except QueueEmpty:
+            logger.debug("Mediator: Bucket queue is empty.")
             pass
         else:
+            logger.debug("Mediator: Running callback for task: %s[%s]" % (
+                task.task_name, task.task_id))
             self.callback(task)
 
 
@@ -96,8 +102,12 @@ class PeriodicWorkController(InfinityThread):
         self.bucket_queue = bucket_queue
 
     def on_iteration(self):
+        logger = get_logger()
+        logger.debug("PeriodicWorkController: Running periodic tasks...")
         self.run_periodic_tasks()
+        logger.debug("PeriodicWorkController: Processing hold queue...")
         self.process_hold_queue()
+        logger.debug("PeriodicWorkController: Going to sleep...")
         time.sleep(1)
 
     def run_periodic_tasks(self):
@@ -106,11 +116,21 @@ class PeriodicWorkController(InfinityThread):
     def process_hold_queue(self):
         """Finds paused tasks that are ready for execution and move
         them to the :attr:`bucket_queue`."""
+        logger = get_logger()
         try:
+            logger.debug(
+                "PeriodicWorkController: Getting next task from hold queue..")
             task, eta = self.hold_queue.get_nowait()
         except QueueEmpty:
+            logger.debug("PeriodicWorkController: Hold queue is empty")
             return
         if datetime.now() >= eta:
+            logger.debug(
+                "PeriodicWorkController: Time to run %s[%s] (%s)..." % (
+                    task.task_name, task.task_id, eta))
             self.bucket_queue.put(task)
         else:
+            logger.debug(
+                "PeriodicWorkController: ETA not ready for %s[%s] (%s)..." % (
+                    task.task_name, task.task_id, eta))
             self.hold_queue.put((task, eta))