added cassandra queue and tests for it

Author: voith

frontera/contrib/backends/__init__.py

@@ -93,11 +93,11 @@ def finished(self):
 class CommonStorageBackend(CommonBackend):
 
     def _create_queue(self, settings):
-        if not isinstance(self.queue_component, BaseQueue):
+        if not issubclass(self.queue_component, BaseQueue):
             raise TypeError('expected queue_component to '
                             'belong to class: %s, got %s instead' % (type(BaseQueue).__name__,
                                                                      type(self.queue_component).__name__))
-        return self.queue_component(self.session_cls,
+        return self.queue_component(self.session,
                                     self.models['QueueModel'],
                                     settings.get('SPIDER_FEED_PARTITIONS'))
 

frontera/contrib/backends/cassandra/__init__.py

@@ -49,7 +49,7 @@ def __init__(self, manager):
         self._states = States(self.session,
                               self.models['StateModel'],
                               settings.get('STATE_CACHE_SIZE_LIMIT'))
-        # self._queue = self._create_queue(settings)
+        self._queue = self._create_queue(settings)
 
     def frontier_stop(self):
         self.states.flush()
@@ -93,8 +93,7 @@ def strategy_worker(cls, manager):
 
         sync_table(model)
 
-        b._states = States(b.session, model,
-                           settings.get('STATE_CACHE_SIZE_LIMIT'))
+        b._states = States(b.session, model, settings.get('STATE_CACHE_SIZE_LIMIT'))
         return b
 
     @classmethod

frontera/contrib/backends/cassandra/components.py

@@ -9,9 +9,7 @@
 from cachetools import LRUCache
 from cassandra import (OperationTimedOut, ReadFailure, ReadTimeout,
                        WriteFailure, WriteTimeout)
-from cassandra.concurrent import execute_concurrent_with_args
 from cassandra.cqlengine.query import BatchQuery
-from w3lib.util import to_bytes, to_native_str
 
 from frontera.contrib.backends import CreateOrModifyPageMixin
 from frontera.contrib.backends.memory import MemoryStates
@@ -22,6 +20,8 @@
 from frontera.utils.misc import chunks, get_crc32
 from frontera.utils.url import parse_domain_from_url_fast
 
+from w3lib.util import to_bytes, to_native_str
+
 
 def _retry(func):
     def func_wrapper(self, *args, **kwargs):
@@ -122,21 +122,26 @@ def flush(self, force_clear=False):
 
 
 class Queue(BaseQueue):
-    def __init__(self, session, queue_cls, partitions, crawl_id, generate_stats, ordering='default'):
+
+    def __init__(self, session, queue_cls, partitions, ordering='default'):
         self.session = session
         self.queue_model = queue_cls
         self.logger = logging.getLogger("frontera.contrib.backends.cassandra.components.Queue")
         self.partitions = [i for i in range(0, partitions)]
         self.partitioner = Crc32NamePartitioner(self.partitions)
         self.ordering = ordering
+        self.batch = BatchQuery()
 
     def frontier_stop(self):
         pass
 
-    def _order_by(self):
+    def _order_by(self, query):
         if self.ordering == 'created':
-            return "created_at"
-        return "created_at"
+            return query.order_by('created_at')
+        if self.ordering == 'created_desc':
+            return query.order_by('-created_at')
+        return query.order_by('score', 'created_at')  # TODO: remove second parameter,
+        # it's not necessary for proper crawling, but needed for tests
 
     def get_next_requests(self, max_n_requests, partition_id, **kwargs):
         """
@@ -148,53 +153,19 @@ def get_next_requests(self, max_n_requests, partition_id, **kwargs):
         """
         results = []
         try:
-            dequeued_urls = 0
-            cql_ditems = []
-            d_query = self.session.prepare("DELETE FROM queue WHERE crawl = ? AND fingerprint = ? AND partition_id = ? "
-                                           "AND score = ? AND created_at = ?")
-            for item in self.queue_model.objects.filter(crawl=self.crawl_id, partition_id=partition_id).\
-                    order_by("partition_id", "score", self._order_by()).limit(max_n_requests):
-                method = 'GET' if not item.method else item.method
-
-                meta_dict2 = dict((name, getattr(item.meta, name)) for name in dir(item.meta)
-                                  if not name.startswith('__'))
-                # TODO: How the result can be an dict not an object -> Objects get error while encodeing for Message Bus
-                # If I take meta_dict2 direct to Request i get the same error message
-
-                meta_dict = dict()
-                meta_dict["fingerprint"] = meta_dict2["fingerprint"]
-                meta_dict["domain"] = meta_dict2["domain"]
-                meta_dict["origin_is_frontier"] = meta_dict2["origin_is_frontier"]
-                meta_dict["scrapy_callback"] = meta_dict2["scrapy_callback"]
-                meta_dict["scrapy_errback"] = meta_dict2["scrapy_errback"]
-                meta_dict["scrapy_meta"] = meta_dict2["scrapy_meta"]
-                meta_dict["score"] = meta_dict2["score"]
-                meta_dict["jid"] = meta_dict2["jid"]
-
-                r = Request(item.url, method=method, meta=meta_dict, headers=item.headers, cookies=item.cookies)
-                r.meta['fingerprint'] = item.fingerprint
-                r.meta['score'] = item.score
+            for item in self._order_by(self.queue_model.filter(partition_id=partition_id).allow_filtering()).limit(max_n_requests):
+                method = item.method or b'GET'
+                r = Request(item.url, method=method, meta=item.meta, headers=item.headers, cookies=item.cookies)
+                r.meta[b'fingerprint'] = to_bytes(item.fingerprint)
+                r.meta[b'score'] = item.score
                 results.append(r)
-
-                cql_d = (item.crawl, item.fingerprint, item.partition_id, item.score, item.created_at)
-                cql_ditems.append(cql_d)
-                dequeued_urls += 1
-
-            if dequeued_urls > 0:
-                execute_concurrent_with_args(self.session, d_query, cql_ditems, concurrency=200)
-
-            self.counter_cls.cass_count({"dequeued_urls": dequeued_urls})
-
+                item.batch(self.batch).delete()
+            self.batch.execute()
         except Exception as exc:
             self.logger.exception(exc)
-
         return results
 
     def schedule(self, batch):
-        query = self.session.prepare("INSERT INTO queue (id, fingerprint, score, partition_id, host_crc32, url, "
-                                     "created_at, meta, depth, headers, method, cookies) "
-                                     "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
-        cql_items = []
         for fprint, score, request, schedule in batch:
             if schedule:
                 _, hostname, _, _, _, _ = parse_domain_from_url_fast(request.url)
@@ -205,35 +176,23 @@ def schedule(self, batch):
                 else:
                     partition_id = self.partitioner.partition(hostname, self.partitions)
                     host_crc32 = get_crc32(hostname)
-                created_at = time()*1E+6
-
-                if "domain" not in request.meta:
-                    request.meta["domain"] = {}
-                if "origin_is_frontier" not in request.meta:
-                    request.meta["origin_is_frontier"] = ''
-                if "scrapy_callback" not in request.meta:
-                    request.meta["scrapy_callback"] = None
-                if "scrapy_errback" not in request.meta:
-                    request.meta["scrapy_errback"] = None
-                if "scrapy_meta" not in request.meta:
-                    request.meta["scrapy_meta"] = {}
-                if "score" not in request.meta:
-                    request.meta["score"] = 0
-                if "jid" not in request.meta:
-                    request.meta["jid"] = 0
-
-                cql_i = (uuid.uuid4(), fprint, score, partition_id, host_crc32, request.url, created_at,
-                         request.meta, 0, request.headers, request.method, request.cookies)
-                cql_items.append(cql_i)
-
-                request.meta['state'] = States.QUEUED
-
-        execute_concurrent_with_args(self.session, query, cql_items, concurrency=400)
-        self.counter_cls.cass_count({"queued_urls": len(cql_items)})
+                q = self.queue_model(id=uuid.uuid4(),
+                                     fingerprint=to_native_str(fprint),
+                                     score=score,
+                                     url=request.url,
+                                     meta=request.meta,
+                                     headers=request.headers,
+                                     cookies=request.cookies,
+                                     method=to_native_str(request.method),
+                                     partition_id=partition_id,
+                                     host_crc32=host_crc32,
+                                     created_at=time() * 1E+6)
+                q.batch(self.batch).save()
+                request.meta[b'state'] = States.QUEUED
+        self.batch.execute()
 
     def count(self):
-        count = self.queue_model.objects.filter().count()
-        return count
+        return self.queue_model.all().count()
 
 
 class BroadCrawlingQueue(Queue):
@@ -265,12 +224,11 @@ def get_next_requests(self, max_n_requests, partition_id, **kwargs):
         while tries < self.GET_RETRIES:
             tries += 1
             limit *= 5.5 if tries > 1 else 1.0
-            self.logger.debug("Try %d, limit %d, last attempt: requests %d, hosts %d" %
-                              (tries, limit, count, len(queue.keys())))
+            self.logger.debug("Try %d, limit %d, last attempt: requests %d, hosts %d",
+                              tries, limit, count, len(queue.keys()))
             queue.clear()
             count = 0
-            for item in self.queue_model.objects.filter(crawl=self.crawl_id, partition_id=partition_id).\
-                    order_by("crawl", "score", self._order_by()).limit(limit):
+            for item in self._order_by(self.queue_model.filter(partition_id=partition_id)).limit(max_n_requests):
                 if item.host_crc32 not in queue:
                     queue[item.host_crc32] = []
                 if max_requests_per_host is not None and len(queue[item.host_crc32]) > max_requests_per_host:
@@ -284,13 +242,14 @@ def get_next_requests(self, max_n_requests, partition_id, **kwargs):
             if min_requests is not None and count < min_requests:
                 continue
             break
-        self.logger.debug("Finished: tries %d, hosts %d, requests %d" % (tries, len(queue.keys()), count))
+        self.logger.debug("Finished: tries %d, hosts %d, requests %d", tries, len(queue.keys()), count)
 
         results = []
-        for items in queue.itervalues():
+        for items in six.itervalues(queue):
             for item in items:
-                method = 'GET' if not item.method else item.method
-                results.append(Request(item.url, method=method, meta=item.meta, headers=item.headers,
-                                       cookies=item.cookies))
-                item.delete()
+                method = item.method or b'GET'
+                results.append(Request(item.url, method=method,
+                                       meta=item.meta, headers=item.headers, cookies=item.cookies))
+                item.batch(self.batch).delete()
+        self.batch.execute()
         return results

frontera/contrib/backends/cassandra/models.py

@@ -73,17 +73,17 @@ def __repr__(self):
 class QueueModel(Model):
     __table_name__ = 'queue'
 
-    id = UUID(primary_key=True)
     partition_id = Integer(primary_key=True)
-    score = Float(required=True)
+    score = Float(primary_key=True)
+    created_at = BigInt(primary_key=True)
+    id = UUID(primary_key=True)
     url = Text(required=True)
     fingerprint = Text(required=True)
     host_crc32 = Integer(required=True)
     meta = PickleDict()
     headers = PickleDict()
     cookies = PickleDict()
     method = Text()
-    created_at = BigInt(required=True)
     depth = SmallInt()
 
     def __repr__(self):

tests/contrib/backends/cassandra/test_backend_cassandra.py

@@ -33,12 +33,13 @@ def setUp(self):
         self.manager = type('manager', (object,), {})
         self.manager.settings = settings
         self.keyspace = settings.CASSANDRABACKEND_KEYSPACE
+        timeout = settings.CASSANDRABACKEND_REQUEST_TIMEOUT
         cluster = Cluster(hosts, port)
         self.session = cluster.connect()
         self.session.execute("CREATE KEYSPACE IF NOT EXISTS %s WITH "
-                             "replication = {'class':'SimpleStrategy', 'replication_factor' : 1}" % self.keyspace)
+                             "replication = {'class':'SimpleStrategy', 'replication_factor' : 1}" % self.keyspace,
+                             timeout=timeout)
         self.session.set_keyspace(self.keyspace)
-        timeout = settings.CASSANDRABACKEND_REQUEST_TIMEOUT
         connection.setup(hosts, self.keyspace, port=port)
         self.session.default_timeout = connection.session.default_timeout = timeout
 
@@ -110,7 +111,7 @@ def assert_db_values(self, model, _filter, fields):
         sync_table(model)
         m = model(**fields)
         m.save()
-        stored_obj = m.get(**_filter)
+        stored_obj = m.objects.allow_filtering().get(**_filter)
         for field, original_value in six.iteritems(fields):
             stored_value = getattr(stored_obj, field)
             if isinstance(original_value, dict):
@@ -198,3 +199,21 @@ def test_state(self):
         self.assertEqual(r4.meta[b'state'], States.CRAWLED)
         state.flush(True)
         self.assertEqual(state._cache, {})
+
+    def test_queue(self):
+        self.manager.settings.SPIDER_FEED_PARTITIONS = 2
+        b = CassandraBackend(self.manager)
+        queue = b.queue
+        batch = [('10', 0.5, r1, True), ('11', 0.6, r2, True),
+                 ('12', 0.7, r3, True)]
+        queue.schedule(batch)
+        self.assertEqual(set([r.url for r in queue.get_next_requests(10, 0,
+                                                                     min_requests=3,
+                                                                     min_hosts=1,
+                                                                     max_requests_per_host=10)]),
+                         set([r3.url]))
+        self.assertEqual(set([r.url for r in queue.get_next_requests(10, 1,
+                                                                     min_requests=3,
+                                                                     min_hosts=1,
+                                                                     max_requests_per_host=10)]),
+                         set([r1.url, r2.url]))