Merge branch 'piotrmaslanka/new-cassandra-backend'

CONTRIBUTORS.txt

@@ -191,3 +191,4 @@ Frantisek Holop, 2015/05/21
 Feanil Patel, 2015/05/21
 Jocelyn Delalande, 2015/06/03
 Juan Rossi, 2015/08/10
+Piotr Maślanka, 2015/08/24

celery/backends/__init__.py

@@ -30,6 +30,7 @@ BACKEND_ALIASES = {
     'db': 'celery.backends.database:DatabaseBackend',
     'database': 'celery.backends.database:DatabaseBackend',
     'cassandra': 'celery.backends.cassandra:CassandraBackend',
+    'new_cassandra': 'celery.backends.new_cassandra:NewCassandraBackend',
     'couchbase': 'celery.backends.couchbase:CouchBaseBackend',
     'couchdb': 'celery.backends.couchdb:CouchDBBackend',
     'riak': 'celery.backends.riak:RiakBackend',

celery/backends/cassandra.py

@@ -17,6 +17,7 @@ except ImportError:  # pragma: no cover
 
 import socket
 import time
+import warnings
 
 from celery import states
 from celery.exceptions import ImproperlyConfigured
@@ -98,6 +99,9 @@ class CassandraBackend(BaseBackend):
 
         self._column_family = None
 
+        warnings.warn("cassandra backend is deprecated. Use new_cassandra instead.",
+                      DeprecationWarning)
+
     def _retry_on_error(self, fun, *args, **kwargs):
         ts = monotonic() + self._retry_timeout
         while 1:

celery/backends/new_cassandra.py

@@ -0,0 +1,188 @@
+# -* coding: utf-8 -*-
+"""
+    celery.backends.new_cassandra
+    ~~~~~~~~~~~~~~~~~~~~~~~~~
+
+    Apache Cassandra result store backend using DataStax driver
+
+"""
+from __future__ import absolute_import
+
+import sys
+try:  # pragma: no cover
+    import cassandra
+except ImportError:  # pragma: no cover
+    cassandra = None   # noqa
+
+from celery import states
+from celery.exceptions import ImproperlyConfigured
+from celery.utils.log import get_logger
+from .base import BaseBackend
+
+__all__ = ['NewCassandraBackend']
+
+logger = get_logger(__name__)
+
+
+class NewCassandraBackend(BaseBackend):
+    """New Cassandra backend utilizing DataStax driver
+
+    .. attribute:: servers
+
+        List of Cassandra servers with format: ``hostname``
+
+    :raises celery.exceptions.ImproperlyConfigured: if
+        module :mod:`cassandra` is not available.
+
+    """
+    supports_autoexpire = True      # autoexpire supported via entry_ttl
+
+    def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,
+                 port=9042, **kwargs):
+        """Initialize Cassandra backend.
+
+        Raises :class:`celery.exceptions.ImproperlyConfigured` if
+        the :setting:`CASSANDRA_SERVERS` setting is not set.
+
+        """
+        super(NewCassandraBackend, self).__init__(**kwargs)
+
+        if not cassandra:
+            raise ImproperlyConfigured(
+                'You need to install the cassandra library to use the '
+                'Cassandra backend. See https://github.com/datastax/python-driver')
+
+        conf = self.app.conf
+        self.servers = (servers or
+                        conf.get('CASSANDRA_SERVERS', None))
+        self.port = (port or
+                     conf.get('CASSANDRA_PORT', None))
+        self.keyspace = (keyspace or
+                         conf.get('CASSANDRA_KEYSPACE', None))
+        self.table = (table or
+                      conf.get('CASSANDRA_TABLE', None))
+
+        if not self.servers or not self.keyspace or not self.table:
+            raise ImproperlyConfigured('Cassandra backend not configured.')
+
+        expires = (entry_ttl or conf.get('CASSANDRA_ENTRY_TTL', None))
+
+        if expires is not None:
+            self.cqlexpires = ' USING TTL %s' % (expires, )
+        else:
+            self.cqlexpires = ''
+
+        read_cons = conf.get('CASSANDRA_READ_CONSISTENCY') or 'LOCAL_QUORUM'
+        write_cons = conf.get('CASSANDRA_WRITE_CONSISTENCY') or 'LOCAL_QUORUM'
+
+        self.read_consistency = getattr(cassandra.ConsistencyLevel,
+            read_cons, cassandra.ConsistencyLevel.LOCAL_QUORUM)
+        self.write_consistency = getattr(cassandra.ConsistencyLevel,
+            write_cons, cassandra.ConsistencyLevel.LOCAL_QUORUM)
+
+        self._connection = None
+        self._session = None
+        self._write_stmt = None
+        self._read_stmt = None
+
+    def process_cleanup(self):
+        if self._connection is not None:
+            self._session.shutdown()
+            self._connection = None
+            self._session = None
+
+    def _get_connection(self, write=False):
+        """
+        Prepare the connection for action
+
+        :param write: bool - are we a writer?
+        """
+        if self._connection is None:
+            self._connection = cassandra.cluster.Cluster(self.servers,
+                                                         port=self.port)
+            self._session = self._connection.connect(self.keyspace)
+
+            # We are forced to do concatenation below, as formatting would
+            # blow up on superficial %s that will be processed by Cassandra
+            self._write_stmt = cassandra.query.SimpleStatement(
+                'INSERT INTO '+self.table+''' (task_id, status, result,'''
+                ''' date_done, traceback, children) VALUES'''
+                ' (%s, %s, %s, %s, %s, %s) '+self.cqlexpires+';')
+            self._write_stmt.consistency_level = self.write_consistency
+
+            self._read_stmt = cassandra.query.SimpleStatement(
+                '''SELECT status, result, date_done, traceback, children
+                   FROM '''+self.table+'''
+                   WHERE task_id=%s LIMIT 1''')
+            self._read_stmt.consistency_level = self.read_consistency
+
+            if write:
+                # Only possible writers "workers" are allowed to issue
+                # CREATE TABLE. This is to prevent conflicting situations
+                # where both task-creator and task-executor would issue it
+                # at the same time.
+
+                # Anyway, if you are doing anything critical, you should
+                # have probably created this table in advance, in which case
+                # this query will be a no-op (instant fail with AlreadyExists)
+                self._make_stmt = cassandra.query.SimpleStatement(
+                    '''CREATE TABLE '''+self.table+''' (
+                        task_id text,
+                        status text,
+                        result blob,
+                        date_done timestamp,
+                        traceback blob,
+                        children blob,
+                        PRIMARY KEY ((task_id), date_done)
+                    )
+                    WITH CLUSTERING ORDER BY (date_done DESC);''')
+                self._make_stmt.consistency_level = self.write_consistency
+                try:
+                    self._session.execute(self._make_stmt)
+                except cassandra.AlreadyExists:
+                    pass
+
+    def _store_result(self, task_id, result, status,
+                      traceback=None, request=None, **kwargs):
+        """Store return value and status of an executed task."""
+        self._get_connection(write=True)
+
+        if sys.version_info >= (3,):
+            buf = lambda x: bytes(x, 'utf8')
+        else:
+            buf = buffer
+
+        self._session.execute(self._write_stmt, (
+            task_id,
+            status,
+            buf(self.encode(result)),
+            self.app.now(),
+            buf(self.encode(traceback)),
+            buf(self.encode(self.current_task_children(request)))
+        ))
+
+    def _get_task_meta_for(self, task_id):
+        """Get task metadata for a task by id."""
+        self._get_connection()
+
+        res = self._session.execute(self._read_stmt, (task_id, ))
+        if not res:
+            return {'status': states.PENDING, 'result': None}
+
+        status, result, date_done, traceback, children = res[0]
+
+        return self.meta_from_decoded({
+            'task_id': task_id,
+            'status': status,
+            'result': self.decode(result),
+            'date_done': date_done.strftime('%Y-%m-%dT%H:%M:%SZ'),
+            'traceback': self.decode(traceback),
+            'children': self.decode(children),
+        })
+
+    def __reduce__(self, args=(), kwargs={}):
+        kwargs.update(
+            dict(servers=self.servers,
+                 keyspace=self.keyspace,
+                 table=self.table))
+        return super(NewCassandraBackend, self).__reduce__(args, kwargs)

celery/tests/backends/test_new_cassandra.py

@@ -0,0 +1,102 @@
+from __future__ import absolute_import
+from pickle import loads, dumps
+from datetime import datetime
+
+from celery import states
+from celery.exceptions import ImproperlyConfigured
+from celery.tests.case import (
+    AppCase, Mock, mock_module, depends_on_current_app
+)
+
+class Object(object):
+    pass
+
+
+class test_NewCassandraBackend(AppCase):
+
+    def setup(self):
+        self.app.conf.update(
+            CASSANDRA_SERVERS=['example.com'],
+            CASSANDRA_KEYSPACE='celery',
+            CASSANDRA_TABLE='task_results',
+        )
+
+    def test_init_no_cassandra(self):
+        """
+        Tests behaviour when no python-driver is installed.
+        new_cassandra should raise ImproperlyConfigured
+        """
+        with mock_module('cassandra'):
+            from celery.backends import new_cassandra as mod
+            prev, mod.cassandra = mod.cassandra, None
+            try:
+                with self.assertRaises(ImproperlyConfigured):
+                    mod.NewCassandraBackend(app=self.app)
+            finally:
+                mod.cassandra = prev
+
+    def test_init_with_and_without_LOCAL_QUROM(self):
+        with mock_module('cassandra'):
+            from celery.backends import new_cassandra as mod
+            mod.cassandra = Mock()
+            cons = mod.cassandra.ConsistencyLevel = Object()
+            cons.LOCAL_QUORUM = 'foo'
+
+            self.app.conf.CASSANDRA_READ_CONSISTENCY = 'LOCAL_FOO'
+            self.app.conf.CASSANDRA_WRITE_CONSISTENCY = 'LOCAL_FOO'
+
+            mod.NewCassandraBackend(app=self.app)
+            cons.LOCAL_FOO = 'bar'
+            mod.NewCassandraBackend(app=self.app)
+
+            # no servers raises ImproperlyConfigured
+            with self.assertRaises(ImproperlyConfigured):
+                self.app.conf.CASSANDRA_SERVERS = None
+                mod.NewCassandraBackend(
+                    app=self.app, keyspace='b', column_family='c',
+                )
+
+    @depends_on_current_app
+    def test_reduce(self):
+        with mock_module('cassandra'):
+            from celery.backends.new_cassandra import NewCassandraBackend
+            self.assertTrue(loads(dumps(NewCassandraBackend(app=self.app))))
+
+    def test_get_task_meta_for(self):
+        with mock_module('cassandra'):
+            from celery.backends import new_cassandra as mod
+            mod.cassandra = Mock()
+            x = mod.NewCassandraBackend(app=self.app)
+            x._connection = True
+            session = x._session = Mock()
+            execute = session.execute = Mock()
+            execute.return_value = [
+                [states.SUCCESS, '1', datetime.now(), b'', b'']
+            ]
+            x.decode = Mock()
+            meta = x._get_task_meta_for('task_id')
+            self.assertEqual(meta['status'], states.SUCCESS)
+
+            x._session.execute.return_value = []
+            meta = x._get_task_meta_for('task_id')
+            self.assertEqual(meta['status'], states.PENDING)
+
+    def test_store_result(self):
+        with mock_module('cassandra'):
+            from celery.backends import new_cassandra as mod
+            mod.cassandra = Mock()
+
+            x = mod.NewCassandraBackend(app=self.app)
+            x._connection = True
+            session = x._session = Mock()
+            session.execute = Mock()
+            x._store_result('task_id', 'result', states.SUCCESS)
+
+    def test_process_cleanup(self):
+        with mock_module('cassandra'):
+            from celery.backends import new_cassandra as mod
+            x = mod.NewCassandraBackend(app=self.app)
+            x.process_cleanup()
+
+            self.assertIsNone(x._connection)
+            self.assertIsNone(x._session)

docs/configuration.rst

@@ -213,6 +213,10 @@ Can be one of the following:
     Use `Cassandra`_ to store the results.
     See :ref:`conf-cassandra-result-backend`.
 
+* new_cassandra
+    Use `new_cassandra`_ to store the results, using newer database driver than _cassandra_.
+    See :ref:`conf-new_cassandra-result-backend`.
+
 * ironcache
     Use `IronCache`_ to store the results.
     See :ref:`conf-ironcache-result-backend`.
@@ -531,7 +535,96 @@ Example configuration
         'taskmeta_collection': 'my_taskmeta_collection',
     }
 
-.. _conf-cassandra-result-backend:
+.. _conf-new_cassandra-result-backend:
+
+
+new_cassandra backend settings
+--------------------------
+
+.. note::
+
+    This Cassandra backend driver requires :mod:`cassandra-driver`.
+    https://pypi.python.org/pypi/cassandra-driver
+
+    To install, use `pip` or `easy_install`:
+
+    .. code-block:: bash
+
+        $ pip install cassandra-driver
+
+This backend requires the following configuration directives to be set.
+
+.. setting:: CASSANDRA_SERVERS
+
+CASSANDRA_SERVERS
+~~~~~~~~~~~~~~~~~
+
+List of ``host`` Cassandra servers. e.g.::
+
+    CASSANDRA_SERVERS = ['localhost']
+
+
+.. setting:: CASSANDRA_PORT
+
+CASSANDRA_PORT
+~~~~~~~~~~~~~~
+
+Port to contact the Cassandra servers on. Default is 9042.
+
+.. setting:: CASSANDRA_KEYSPACE
+
+CASSANDRA_KEYSPACE
+~~~~~~~~~~~~~~~~~~
+
+The keyspace in which to store the results. e.g.::
+
+    CASSANDRA_KEYSPACE = 'tasks_keyspace'
+
+.. setting:: CASSANDRA_COLUMN_FAMILY
+
+CASSANDRA_TABLE
+~~~~~~~~~~~~~~~~~~~~~~~
+
+The table (column family) in which to store the results. e.g.::
+
+    CASSANDRA_TABLE = 'tasks'
+
+.. setting:: CASSANDRA_READ_CONSISTENCY
+
+CASSANDRA_READ_CONSISTENCY
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The read consistency used. Values can be ``ONE``, ``TWO``, ``THREE``, ``QUORUM``, ``ALL``,
+``LOCAL_QUORUM``, ``EACH_QUORUM``, ``LOCAL_ONE``.
+
+.. setting:: CASSANDRA_WRITE_CONSISTENCY
+
+CASSANDRA_WRITE_CONSISTENCY
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The write consistency used. Values can be ``ONE``, ``TWO``, ``THREE``, ``QUORUM``, ``ALL``,
+``LOCAL_QUORUM``, ``EACH_QUORUM``, ``LOCAL_ONE``.
+
+.. setting:: CASSANDRA_ENTRY_TTL
+
+CASSANDRA_ENTRY_TTL
+~~~~~~~~~~~~~~~~~~~
+
+Time-to-live for status entries. They will expire and be removed after that many seconds
+after adding. Default (None) means they will never expire.
+
+Example configuration
+~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: python
+
+    CASSANDRA_SERVERS = ['localhost']
+    CASSANDRA_KEYSPACE = 'celery'
+    CASSANDRA_COLUMN_FAMILY = 'task_results'
+    CASSANDRA_READ_CONSISTENCY = 'ONE'
+    CASSANDRA_WRITE_CONSISTENCY = 'ONE'
+    CASSANDRA_ENTRY_TTL = 86400
+
 
 Cassandra backend settings
 --------------------------

docs/includes/installation.txt

@@ -78,7 +78,10 @@ Transports and Backends
     for using memcached as a result backend.
 
 :celery[cassandra]:
-    for using Apache Cassandra as a result backend.
+    for using Apache Cassandra as a result backend with pycassa driver.
+
+:celery[new_cassandra]:
+    for using Apache Cassandra as a result backend with DataStax driver.
 
 :celery[couchdb]:
     for using CouchDB as a message transport (*experimental*).

docs/internals/reference/celery.backends.new_cassandra.rst

@@ -0,0 +1,11 @@
+================================================
+ celery.backends.new_cassandra
+================================================
+
+.. contents::
+    :local:
+.. currentmodule:: celery.backends.new_cassandra
+
+.. automodule:: celery.backends.new_cassandra
+    :members:
+    :undoc-members:

docs/whatsnew-3.2.rst

@@ -99,6 +99,11 @@ Bla bla
 
 - blah blah
 
+New Cassandra Backend
+=====================
+New Cassandra backend will be called new_cassandra and utilize python-driver.
+Old backend is now deprecated.
+
 
 Event Batching
 ==============

requirements/extras/new_cassandra.txt

@@ -0,0 +1 @@
+cassandra-driver

setup.py

@@ -177,6 +177,7 @@ features = {
     'eventlet', 'gevent', 'msgpack', 'yaml', 'redis',
     'mongodb', 'sqs', 'couchdb', 'riak', 'beanstalk', 'zookeeper',
     'zeromq', 'sqlalchemy', 'librabbitmq', 'pyro', 'slmq',
+    'new_cassandra',
 }
 extras_require = {x: extras(x + '.txt') for x in features}
 extra['extras_require'] = extras_require