Logging colors: Fixes 'mod_wsgi.Log object has no 'isatty' attribute'

celery/conf.py

@@ -80,6 +80,12 @@ _DEFAULTS = {
 }
 
 
+def isatty(fh):
+    # Fixes bug with mod_wsgi:
+    #   mod_wsgi.Log object has no attribute isatty.
+    return getattr(fh, "isatty", None) and fh.isatty()
+
+
 _DEPRECATION_FMT = """
 %s is deprecated in favor of %s and is scheduled for removal in celery v1.4.
 """.strip()
@@ -139,8 +145,8 @@ CELERYD_LOG_FORMAT = _get("CELERYD_LOG_FORMAT",
                           compat=["CELERYD_DAEMON_LOG_FORMAT"])
 CELERYD_TASK_LOG_FORMAT = _get("CELERYD_TASK_LOG_FORMAT")
 CELERYD_LOG_FILE = _get("CELERYD_LOG_FILE")
-CELERYD_LOG_COLOR = _get("CELERYD_LOG_COLOR", \
-                       CELERYD_LOG_FILE is None and sys.stderr.isatty())
+CELERYD_LOG_COLOR = _get("CELERYD_LOG_COLOR",
+                       CELERYD_LOG_FILE is None and isatty(sys.stderr))
 CELERYD_LOG_LEVEL = _get("CELERYD_LOG_LEVEL",
                             compat=["CELERYD_DAEMON_LOG_LEVEL"])
 CELERYD_LOG_LEVEL = LOG_LEVELS[CELERYD_LOG_LEVEL.upper()]

celery/events/state.py

@@ -1,7 +1,12 @@
+import time
+import heapq
+
 from carrot.utils import partition
 
 from celery import states
 
+HEARTBEAT_EXPIRE = 150 # 2 minutes, 30 seconds
+
 
 class Thing(object):
     visited = False
@@ -16,21 +21,27 @@ class Thing(object):
 
 
 class Worker(Thing):
-    alive = False
 
     def __init__(self, **fields):
         super(Worker, self).__init__(**fields)
         self.heartbeats = []
 
-    def online(self, **kwargs):
-        self.alive = True
+    def online(self, timestamp=None, **kwargs):
+        self._heartpush(timestamp)
 
     def offline(self, **kwargs):
-        self.alive = False
+        self.heartbeats = []
 
     def heartbeat(self, timestamp=None, **kwargs):
-        self.heartbeats.append(timestamp)
-        self.alive = True
+        self._heartpush(timestamp)
+
+    def _heartpush(self, timestamp):
+        heapq.heappush(self.heartbeats, timestamp)
+
+    @property
+    def alive(self):
+        return (self.heartbeats and
+                time.time() < self.heartbeats[0] + HEARTBEAT_EXPIRE)
 
 
 class Task(Thing):

celery/log.py

@@ -24,6 +24,7 @@ COLOURS = {
     'ERROR': RED
 }
 
+
 class ColourFormatter(logging.Formatter):
     def __init__(self, msg, use_colour=True):
         logging.Formatter.__init__(self, msg)
@@ -32,9 +33,11 @@ class ColourFormatter(logging.Formatter):
     def format(self, record):
         levelname = record.levelname
         if self.use_colour and levelname in COLOURS:
-            record.msg = COLOUR_SEQ % (30 + COLOURS[levelname]) + record.msg + RESET_SEQ
+            record.msg = COLOUR_SEQ % (
+                    30 + COLOURS[levelname]) + record.msg + RESET_SEQ
         return logging.Formatter.format(self, record)
 
+
 def get_task_logger(loglevel=None):
     ensure_process_aware_logger()
     logger = logging.getLogger("celery.Task")