Merge branch 'veezio/cassandra_detailed'

celery/backends/cassandra.py

@@ -33,11 +33,12 @@ class CassandraBackend(BaseDictBackend):
     servers = []
     keyspace = None
     column_family = None
+    detailed_mode = False
     _retry_timeout = 300
     _retry_wait = 3
 
     def __init__(self, servers=None, keyspace=None, column_family=None,
-            cassandra_options=None, **kwargs):
+            cassandra_options=None, detailed_mode=False, **kwargs):
         """Initialize Cassandra backend.
 
         Raises :class:`celery.exceptions.ImproperlyConfigured` if
@@ -67,6 +68,9 @@ class CassandraBackend(BaseDictBackend):
         self.cassandra_options = dict(cassandra_options or {},
                                    **self.app.conf.get("CASSANDRA_OPTIONS",
                                                        {}))
+        self.detailed_mode = detailed_mode or \
+                                self.app.conf.get("CASSANDRA_DETAILED_MODE",
+                                                  self.detailed_mode)
         read_cons = self.app.conf.get("CASSANDRA_READ_CONSISTENCY",
                                       "LOCAL_QUORUM")
         write_cons = self.app.conf.get("CASSANDRA_WRITE_CONSISTENCY",
@@ -126,11 +130,16 @@ class CassandraBackend(BaseDictBackend):
             cf = self._get_column_family()
             date_done = self.app.now()
             meta = {"status": status,
-                    "result": self.encode(result),
                     "date_done": date_done.strftime('%Y-%m-%dT%H:%M:%SZ'),
                     "traceback": self.encode(traceback)}
-            cf.insert(task_id, meta,
-                      ttl=timedelta_seconds(self.expires))
+            if self.detailed_mode:
+                meta["result"] = result
+                cf.insert(task_id, {date_done: self.encode(meta)},
+                          ttl=timedelta_seconds(self.expires))
+            else:
+                meta["result"] = self.encode(result)
+                cf.insert(task_id, meta,
+                          ttl=timedelta_seconds(self.expires))
 
         return self._retry_on_error(_do_store)
 
@@ -140,14 +149,19 @@ class CassandraBackend(BaseDictBackend):
         def _do_get():
             cf = self._get_column_family()
             try:
-                obj = cf.get(task_id)
-                meta = {
-                    "task_id": task_id,
-                    "status": obj["status"],
-                    "result": self.decode(obj["result"]),
-                    "date_done": obj["date_done"],
-                    "traceback": self.decode(obj["traceback"]),
-                }
+                if self.detailed_mode:
+                    row = cf.get(task_id, column_reversed=True, column_count=1)
+                    meta = self.decode(row.values()[0])
+                    meta["task_id"] = task_id
+                else:
+                    obj = cf.get(task_id)
+                    meta = {
+                        "task_id": task_id,
+                        "status": obj["status"],
+                        "result": self.decode(obj["result"]),
+                        "date_done": obj["date_done"],
+                        "traceback": self.decode(obj["traceback"]),
+                    }
             except (KeyError, pycassa.NotFoundException):
                 meta = {"status": states.PENDING, "result": None}
             return meta

docs/configuration.rst

@@ -199,6 +199,10 @@ Can be one of the following:
     Send results back as AMQP messages
     See :ref:`conf-amqp-result-backend`.
 
+* cassandra
+    Use `Cassandra`_ to store the results.
+    See :ref:`conf-cassandra-result-backend`.
+
 .. warning:
 
     While the AMQP result backend is very efficient, you must make sure
@@ -209,6 +213,7 @@ Can be one of the following:
 .. _`MongoDB`: http://mongodb.org
 .. _`Redis`: http://code.google.com/p/redis/
 .. _`Tokyo Tyrant`: http://1978th.net/tokyotyrant/
+.. _`Cassandra`: http://cassandra.apache.org/
 
 .. setting:: CELERY_RESULT_SERIALIZER
 
@@ -528,6 +533,82 @@ Example configuration
         "taskmeta_collection": "my_taskmeta_collection",
     }
 
+.. _conf-cassandra-result-backend:
+
+Cassandra backend settings
+--------------------------
+
+.. note::
+
+    The Cassandra backend requires the :mod:`pycassa` library:
+    http://pypi.python.org/pypi/pycassa/
+
+    To install the pycassa package use `pip` or `easy_install`::
+
+        $ pip install pycassa
+
+This backend requires the following configuration directives to be set.
+
+.. setting:: CASSANDRA_SERVERS
+
+CASSANDRA_SERVERS
+~~~~~~~~~~~~~~~~~
+
+List of `host:port` Cassandra servers. e.g. `["localhost:9160]"`.
+
+.. setting:: CASSANDRA_KEYSPACE
+
+CASSANDRA_KEYSPACE
+~~~~~~~~~~~~~~~~~~
+
+The keyspace in which to store the results. e.g. `"tasks_keyspace"`.
+
+.. setting:: CASSANDRA_COLUMN_FAMILY
+
+CASSANDRA_COLUMN_FAMILY
+~~~~~~~~~~~~~~~~~~~~~~~
+
+The column family in which to store the results. eg `"tasks"`
+
+.. setting:: CASSANDRA_READ_CONSISTENCY
+
+CASSANDRA_READ_CONSISTENCY
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The read consistency used. Values can be `"ONE"`, `"QUORUM"` or `"ALL"`.
+
+.. setting:: CASSANDRA_WRITE_CONSISTENCY
+
+CASSANDRA_WRITE_CONSISTENCY
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The write consistency used. Values can be `"ONE"`, `"QUORUM"` or `"ALL"`.
+
+.. setting:: CASSANDRA_WRITE_CONSISTENCY
+
+CASSANDRA_DETAILED_MODE
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Enable or disable detailed mode. Default is `"False"`.
+This mode allows to use the power of Cassandra wide columns to store all states for a task as a wide column, instead of only the last one.
+
+To use this mode, you need to configure your ColumnFamily to use the `TimeUUID` type as a comparator::
+
+    create column family task_results with comparator = TimeUUIDType;
+
+Example configuration
+~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: python
+
+    CASSANDRA_SERVERS = ["localhost:9160"]
+    CASSANDRA_KEYSPACE = "celery"
+    CASSANDRA_COLUMN_FAMILY = "task_results"
+    CASSANDRA_READ_CONSISTENCY = "ONE"
+    CASSANDRA_WRITE_CONSISTENCY = "ONE"
+    CASSANDRA_DETAILED_MODE = True
+
+
 .. _conf-messaging:
 
 Message Routing