Changing waiting logic

workers/reindexer.py

@@ -16,7 +16,7 @@ class NotEmptyQueueException(Exception):
     pass
 
 
-def create_sampling_task(cfg, uuid):
+def create_sampling_task(cfg, channel, uuid):
 
     # here-below we generate a task for the sample generator (full -> meta)
     msg = dict()
@@ -27,8 +27,8 @@ def create_sampling_task(cfg, uuid):
 
     the_body = msgpack.packb(msg, use_bin_type=True)
 
-    connection = pika.BlockingConnection(pika.ConnectionParameters(host=cfg['rabbitmq']['host']))
-    channel = connection.channel()
+    # connection = pika.BlockingConnection(pika.ConnectionParameters(host=cfg['rabbitmq']['host']))
+    # channel = connection.channel()
     exchange = cfg['rabbitmq']['exchange']
 
     queue_name  = cfg['rabbitmq']['queue_name_6']
@@ -44,7 +44,7 @@ def create_sampling_task(cfg, uuid):
                            properties=pika.BasicProperties(delivery_mode = 2)
                          )
 
-    connection.close()
+    #connection.close()
 
 
     return
@@ -55,6 +55,7 @@ def on_msg_callback(channel, method, properties, body):
     decoded_body = msgpack.unpackb(body, raw=False)
     cfg = decoded_body['header']['cfg']
     uuid = decoded_body['body']
+    count_ref = decoded_body['header']['count']
 
     # from lib.elasticsearch_template import template
     # template['index_patterns'] = [ cfg['reindexer']['destination_index'] ]
@@ -74,40 +75,49 @@ def on_msg_callback(channel, method, properties, body):
     the_query['query']['term'] = {'uuid.keyword': '{0}'.format(uuid)}
 
     es_source.indices.refresh(index=cfg['reindexer']['source_index'])
-    count1 = es_source.count(cfg['reindexer']['source_index'], body=the_query).get('count')
-    logging.debug("%i document(s) found in the source index with uuid = %s" % (count1, uuid))
-
-    if uuid.endswith('.full'):
-
-        logging.debug("Waiting for 30 seconds before counting again...")
-        time.sleep(30)
-
-        es_source.indices.refresh(index=cfg['reindexer']['source_index'])
-        count2 = es_source.count(cfg['reindexer']['source_index'], body=the_query).get('count')
-        logging.debug("%i document(s) found in the source index with uuid = %s" % (count2, uuid))
-
-        if count1 != count2 or count2 == 0:
+    count_es = es_source.count(cfg['reindexer']['source_index'], body=the_query).get('count')
+    # logging.debug("%i document(s) found in the source index with uuid = %s" % (count1, uuid))
 
-            logging.warning('Documents are still being pushed to the source index. Waiting...')
-            time.sleep(5)
-            channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
-            return
-            #raise NotEmptyQueueException('Documents are still being pushed to the source index. Waiting...')
-
-    elif uuid.endswith('.meta'):
-
-        if count1 != 1:
+    # if uuid.endswith('.full'):
+    #
+    #     logging.debug("Waiting for 5 seconds before counting again...")
+    #     time.sleep(5)
+    #
+    #     es_source.indices.refresh(index=cfg['reindexer']['source_index'])
+    #     count2 = es_source.count(cfg['reindexer']['source_index'], body=the_query).get('count')
+    #     logging.debug("%i document(s) found in the source index with uuid = %s" % (count2, uuid))
+    #
+    #     if count1 != count2 or count2 == 0:
+    #
+    #         logging.warning('Documents are still being pushed to the source index. Waiting...')
+    #         time.sleep(5)
+    #         channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
+    #         return
+    #         #raise NotEmptyQueueException('Documents are still being pushed to the source index. Waiting...')
+    #
+    # elif uuid.endswith('.meta'):
+    #
+    #     if count1 != 1:
+    #
+    #         logging.warning('Documents are still being pushed to the source index. Waiting...')
+    #         time.sleep(5)
+    #         channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
+    #         return
+    #
+    # else:
+    #     channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
+    #     logging.error("The uuid ends neither with .full nor with .meta. What shall I do?")
+        # return
 
-            logging.warning('Documents are still being pushed to the source index. Waiting...')
-            time.sleep(5)
-            channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
-            return
 
-    else:
+    if count_es != count_ref:
+        logging.warning('Documents are still being pushed to the source index for dataset with uuid = %s' % uuid)
+        logging.debug('count_es = %i; count_ref = %i' % (count_es, count_ref))
+        time.sleep(5)
         channel.basic_nack(delivery_tag = method.delivery_tag, requeue=1)
-        logging.error("The uuid ends neither with .full nor with .meta. What shall I do?")
         return
 
+
     # 1. remove already existing docs from destination index
     logging.info("Removing dataset with uuid = %s from the destination index..." % uuid)
 
@@ -127,6 +137,8 @@ def on_msg_callback(channel, method, properties, body):
     try:
         res = es.delete_by_query(index, doc_type='_doc', body=the_query)
         logging.debug(res)
+        res = es.indices.refresh(index=index)
+        logging.debug(res)
     except NotFoundError:
         pass
     except Exception as e:
@@ -178,7 +190,7 @@ def on_msg_callback(channel, method, properties, body):
 
         # 3. create sampling task (full -> meta)
         if '.full' in uuid:
-            create_sampling_task(cfg, uuid)#, reindex_task_url)
+            create_sampling_task(cfg, channel, uuid)#, reindex_task_url)
             logging.info("Created sampling task.")